package com.surfilter.massdata.spark.task.domainprocess;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;

import scala.Tuple2;

import com.carrotsearch.hppc.ObjectLookupContainer;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.surfilter.massdata.es.core.client.EsClient;
import com.surfilter.massdata.es.core.client.pool.EsClientPool;
import com.surfilter.massdata.es.core.entity.EsEntity;
import com.surfilter.massdata.es.core.entity.IndexClass;
import com.surfilter.massdata.es.core.service.EsMappingService;
import com.surfilter.massdata.es.core.service.impl.EsMappingServiceImpl;
import com.surfilter.massdata.es.core.util.DateUtils;
import com.surfilter.massdata.spark.bean.massDomain.CdnInvalid;
import com.surfilter.massdata.spark.bean.massDomain.DomainHistory;
import com.surfilter.massdata.spark.bean.massDomain.DomainInvalid;
import com.surfilter.massdata.spark.bean.massDomain.DomainIp;
import com.surfilter.massdata.spark.bean.massDomain.DomainSurvival;
import com.surfilter.massdata.spark.bean.massDomain.FullDomainIndex;
import com.surfilter.massdata.spark.bean.massDomain.TopDomainIndex;
import com.surfilter.massdata.spark.bean.massDomain.WebSiteIpSurvival;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.EsFullDomainVisits;
import com.surfilter.massdata.spark.model.EsTopDomainVisits;
import com.surfilter.massdata.spark.model.StructDomain;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DNSParser;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.DomainDatabase;
import com.surfilter.massdata.spark.util.IPDatabase;
import com.surfilter.massdata.spark.util.JdbcUtil;
import com.surfilter.massdata.spark.util.SysProps;
import com.surfilter.massdata.spark.util.URLUtil;

public class MassDomainTask implements Serializable{
	
	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(MassDomainTask.class);
	
	public static void main(String[] args) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		String ymdhms = "yyyy-MM-dd HH:mm:ss";
		SimpleDateFormat yyyyMMddHHmmss = new SimpleDateFormat(ymdhms);
		System.out.println("MassDomainTask begin:"+yyyyMMddHHmmss.format(new Date())+"........................");
		try{
			SparkConf conf = new SparkConf();
			conf.setAppName("MassDomainTask");
			// 设置序列化器为KryoSerializer。  
			conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
			// 注册要序列化的自定义类型。  
			conf.registerKryoClasses(new Class[]{DomainHistory.class,DomainInvalid.class,DomainSurvival.class,
					DomainIp.class,WebSiteIpSurvival.class,CdnInvalid.class,TopDomainIndex.class,FullDomainIndex.class});

			SysProps sysProps = SysProps.getInstance();
			
			String esNode = sysProps.getValue("es.nodes");
			String esPort = sysProps.getValue("es.port");
	
			conf.set("es.nodes",esNode);
			conf.set("es.port",esPort);
			List<String> surfilterIpList = getSurfilterIp(sysProps);
			String hiveDB = sysProps.getValue("hiveDB");

			JavaSparkContext sc = new JavaSparkContext(conf);
			//HiveContext与SQLContext如果需要同时使用，需要如下方式，若直接new SQLContext对象，各自注册的临时表会找不到
			HiveContext hiveSql = new HiveContext(sc);
			SQLContext sqlContext = hiveSql;
			
			String dayStr = "";
			if(args.length > 0){
				//时间格式:20170412
				dayStr = args[0];
			}
			//默认执行时昨天的数据
			Date date = getExecDate(DateUtil.getYesterday(),dayStr);
			String dateFormat = DateUtil.getCurrentymd(date);
			
			if(log.isInfoEnabled()){
				log.info("es.nodes="+esNode);
				log.info("es.port="+esPort);
				log.info("hiveDB="+hiveDB);
				log.info("surfilter.114.ips="+surfilterIpList);
				log.info("date="+dateFormat);
			}
			
			Broadcast<IPDatabase> ipBroadcast = sc.broadcast(IPDatabase.getInstance());
			
			//1.拨测结果与存活结果处理-->合成中间结果域名IP
			//过滤拨测结果省外IP,114IP
			DataFrame domainInvalidFrame = getDomainInvalidFrame(sqlContext,hiveSql,ipBroadcast,surfilterIpList,dateFormat,hiveDB);
			domainInvalidFrame.registerTempTable("domain_invalid_temp");
			System.out.println("===============domainInvalidFrame ==============="+domainInvalidFrame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("domainInvalidFrame exectime: "+exectime+" min........................");
			
			
			
			//过滤存活结果省外IP,114IP
			DataFrame domainSurvivalFrame = getDomainSurvivalFrame(sqlContext,hiveSql,ipBroadcast,surfilterIpList,dateFormat,hiveDB);
			domainSurvivalFrame.registerTempTable("domain_survival_temp");
			sqlContext.dropTempTable("domain_survival_filter_temp");
			System.out.println("===============domainSurvivalFrame ==============="+domainSurvivalFrame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("domainSurvivalFrame exectime: "+exectime+" min........................");
			
			
			//拨测结果与存活结果关联
			DataFrame domainIpFrame = getDomainIpDataFrame(hiveSql, sqlContext,domainSurvivalFrame,dateFormat,hiveDB,surfilterIpList);
			domainIpFrame.registerTempTable("domain_ip_temp");
			System.out.println("===============domainIpFrame ==============="+domainIpFrame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("domainIpFrame exectime: "+exectime+" min........................");
			
			
			//2.域名IP中间结果 历史域名-->合成中间结果新历史域名V1.O
			//过滤历史域名中IP为省外IP和114IP
			JavaRDD<DomainHistory> domainHistoryRDD = getDomainHistoryRDD(hiveDB, sqlContext,ipBroadcast,surfilterIpList,dateFormat);
			
			//历史域名中相同的域名进行合并
			DataFrame domainHistoryFrame = getMergeDomainHistoryFrame(sqlContext,domainHistoryRDD);
			domainHistoryFrame.registerTempTable("domain_history_temp");
			sqlContext.dropTempTable("domain_history_filter_temp");
			
			//域名IP中间结果与历史域名合并为新历史域名V1.O逻辑
			DataFrame domainHistoryV1Frame = getDomainHistoryV1Frame(sqlContext,dateFormat,hiveDB);
			domainHistoryV1Frame.registerTempTable("domain_history_v1_temp");
			sqlContext.dropTempTable("domain_history_filter_v1_temp");
			sqlContext.dropTempTable("domain_ip_temp");
			System.out.println("===============domainHistoryV1Frame ==============="+domainHistoryV1Frame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("domainHistoryV1Frame exectime: "+exectime+" min........................");
			
			
			
			//3.新历史域名V1.0关联存活结果表-->合成新历史域名V2.0
			DataFrame domainHistroyV2Frame = getDomainHistoryV2Frame(sqlContext);
			domainHistroyV2Frame.registerTempTable("domain_history_v2_temp");
			sqlContext.dropTempTable("domain_history_v1_temp");
			
			//4.新历史域名V2.0与天域名-->合并为中间结果新历史域名V3.0
			//天域名过滤外省IP和114IP
			JavaRDD<DomainIp> dayDomainRDD = getDayDomainRDD(sqlContext,ipBroadcast,surfilterIpList,hiveDB,dateFormat);
			
			//天域名中相同的域名进行合并
			DataFrame dayDomainFrame = getMergeDayDomainFrame(sqlContext,dayDomainRDD);
			dayDomainFrame.registerTempTable("day_domain_temp");
			sqlContext.dropTempTable("day_domain_filter_temp");
			
			//新历史域名V2.0与天域名关联
			DataFrame domainHistoryV3Frame = getDomainHistroyV3Frame(sqlContext,dateFormat,hiveDB);
			sqlContext.dropTempTable("domain_history_v2_temp");
			
			JavaRDD<Row> domainHisV3RDD = domainHistoryV3Frame.toJavaRDD();
			System.out.println("===============domainHistoryV3Frame ==============="+domainHistoryV3Frame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("domainHistoryV3Frame exectime: "+exectime+" min........................");
			
			
			
			//新历史域名V3.0合并逻辑
			JavaRDD<DomainHistory> domainHistoryV3RDD = getDomainHistroyV3RDD(sqlContext, domainHisV3RDD,surfilterIpList);
			
			//将历史域名V3.0内容上传到HDFS
			uploadDomainHistoryV3ToHDFS(domainHisV3RDD,sysProps);
			
			//5.新历史域名V1.0，存活结果及CDN检测结果合并为网站IP存活关系V1.0
			//将新历史域名V1.0中相同的顶级域名进行合并
			DataFrame historyV1MergeDataFrame = geHistoryV1MergeFrame(sqlContext, domainHistoryV1Frame);
			historyV1MergeDataFrame.registerTempTable("domain_history_merge_temp");
			
			//将存活结果中相同的顶级域名进行合并	
			DataFrame domainSurvivalMergeFrame = getDomainSurvivalMergeFrame(sqlContext, domainSurvivalFrame);
			domainSurvivalMergeFrame.registerTempTable("domain_survival_merge_temp");
			
			//将新历史域名V1.0和存活结果合并后的结果进行处理
			DataFrame histroySurvivalMergeFrame = getHistroySurvivalMergeFrame(sqlContext);
			histroySurvivalMergeFrame.registerTempTable("history_survival_merge_temp");
			sqlContext.dropTempTable("domain_history_merge_temp");
			sqlContext.dropTempTable("domain_survival_merge_temp");
			System.out.println("===============histroySurvivalMergeFrame ==============="+histroySurvivalMergeFrame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("histroySurvivalMergeFrame exectime: "+exectime+" min........................");
			
			
			
			//将CDN检测结果中的顶级域名相同进行合并操作
			DataFrame cdnInvalidFrame = getCdnInvalidMergeFrame(hiveSql, sqlContext, ipBroadcast,surfilterIpList,dateFormat,hiveDB);
			cdnInvalidFrame.registerTempTable("domain_cdn_merge_temp");
			
			//将将新历史域名V1.0和存活结果合并后的结果与CDN检测合并后的结果在进行合并
			DataFrame historySurCdnFrame = getHistorySurCdnMergeFrame(sqlContext);
			historySurCdnFrame.registerTempTable("his_sur_cdn_merge_temp");	
			sqlContext.dropTempTable("domain_cdn_merge_temp");
			System.out.println("===============historySurCdnFrame ==============="+historySurCdnFrame.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("historySurCdnFrame exectime: "+exectime+" min........................");
			
			
			
			//6.网站IP存活关系V1.0与天域名合并为临时结果网站IP存活关系V2.0
			//将天域名中顶级域名相同的进行合并操作
			DataFrame dayDomainMergeFrame = getDayDomainMergeFrame(dayDomainFrame,hiveSql,sqlContext,dateFormat,surfilterIpList,hiveDB);
			dayDomainMergeFrame.registerTempTable("day_domain_merge_temp");
			
			//补充天域名中的结果，全链接
			JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD = getDayHisSurCdnMergeRDD(sqlContext);
			sqlContext.dropTempTable("day_domain_merge_temp");
			System.out.println("===============webSiteIpSurvivalRDD==============="+webSiteIpSurvivalRDD.count());
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("webSiteIpSurvivalRDD exectime: "+exectime+" min........................");
			
			
			System.out.println("===============insert BR2006 begin===============");
			//网站IP表入库(BR2006)
			insertBr2006(webSiteIpSurvivalRDD,ipBroadcast);
			System.out.println("=============insert BR2006 end=================");
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("BR2006 exectime: "+exectime+" min........................");
			
			
			System.out.println("===============insert BR2005 begin===============");
			//网站表入库(BR2005)
			insertBr2005(webSiteIpSurvivalRDD);
			System.out.println("===============insert BR2005 end===============");
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("BR2005 exectime: "+exectime+" min........................");
			
			
			System.out.println("===============topdomain insert into dns_xxxx_xx_xx_provincetld of ES begin===============webSiteIpSurvivalRDD:"+webSiteIpSurvivalRDD.count());
			//ES入库在网站表入库之后，因需要用到网站表的数据
			Broadcast<DomainDatabase> domainBroadcast = sc.broadcast(DomainDatabase.getInstance());
			//顶级域名信息入ES
			insertTopDomainToEs(esNode, ipBroadcast, domainBroadcast,webSiteIpSurvivalRDD,dateFormat,sysProps);
			System.out.println("===============topdomain insert into ES end===============");
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("topdomain insert exectime: "+exectime+" min........................");
			
			
			System.out.println("===============all domain insert into  dns_xxxx_xx_xx_ipdomainrelation of ES begin===============domainHistoryV3RDD:"+domainHistoryV3RDD.count());
			//将历史域名V3.0插入全量域名信息到ES中
			System.out.println("es.coalesce_Partitions:"+sysProps.getInteger("es.coalesce_Partitions", 10));
			domainHistoryV3RDD.coalesce(sysProps.getInteger("es.coalesce_Partitions", 10), true);

			insertFullDomainToES(esNode, ipBroadcast,domainBroadcast,domainHistoryV3RDD,dateFormat,sysProps);
			System.out.println("===============all domain insert into ES end===============");
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("all domain inser exectime: "+exectime+" min........................");
			
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		exectime=(end-start)*1.0/(1000*60);
		System.out.println("MassDomainTask exectime: "+exectime+" min........................");
		System.out.println("MassDomainTask end:"+yyyyMMddHHmmss.format(new Date())+"........................");
	}
	
	/**
	 * 过滤省外IP和114IP，IP全部不满足条件的删除该条记录
	 * @param hiveDB
	 * @param sqlContext
	 * @param ipDatabase
	 * @param surfilterIpList
	 * @param date
	 * @return
	 */
	@SuppressWarnings("serial")
	private static JavaRDD<DomainHistory> getDomainHistoryRDD(String hiveDB,SQLContext sqlContext,final Broadcast<IPDatabase> ipBroadcast,final List<String> surfilterIpList,String date) {
		DataFrame df = sqlContext.sql("select domain,ip,timestamps,isEffective,isAlive from "+hiveDB+".domain_history where ds = '"+date+"'");
		JavaRDD<DomainHistory> rdd = df.toJavaRDD().map(new Function<Row, DomainHistory>() {

			@Override
			public DomainHistory call(Row row){
				DomainHistory dh = new DomainHistory();
				try{
					String ip = row.getAs("ip");
					StringBuffer ipBuffer = new StringBuffer();
					//IP过滤
					if(StringUtils.isNotBlank(ip)){
						String[] ipSplit = ip.split("\\|");
						for(String ipStr: ipSplit){
							StructIP structIp = ipBroadcast.value().get(ipStr);
							//过滤外省IP
							if(structIp != null){//为null则为外省
								//过滤114IP
								if(!surfilterIpList.contains(ipStr)){
									ipBuffer.append(ipStr).append("|");
								}
							}
						}
					}
					if(ipBuffer.length() > 0){
						String domain = row.getAs("domain");
						long timestamps = Long.parseLong(row.getAs("timestamps").toString());
						long isEffective = Long.parseLong(row.getAs("isEffective").toString());
						long isAlive = Long.parseLong(row.getAs("isAlive").toString());
						dh.setDomain(domain);
						dh.setIp(getFilter(ipBuffer));
						dh.setTimestamps(timestamps);
						dh.setIsEffective(isEffective);
						dh.setIsAlive(isAlive);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return dh;
			}
		});
		
		return rdd;
	}
	/**
	 * 处理逻辑：相同的域名进行合并，同时合并IP并去重
	 * @param sqlContext
	 * @param rdd
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getMergeDomainHistoryFrame(SQLContext sqlContext,JavaRDD<DomainHistory> rdd) {
		JavaRDD<DomainHistory> domainRDD = rdd.mapToPair(new PairFunction<DomainHistory,String,String>() {

			@Override
			public Tuple2<String,String> call(DomainHistory dh){
				try{
					String domain = dh.getDomain();
					String ip = dh.getIp();
					long timestamps = dh.getTimestamps();
					long isEffective = dh.getIsEffective();
					long isAlive = dh.getIsAlive();
					if(StringUtils.isNotBlank(domain)){
						return new Tuple2<String,String>(domain,ip+"@"+timestamps+"@"+isEffective+"@"+isAlive);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","");
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2) throws Exception {
				return v1 + "@@" + v2;
			}
		}).map(new Function<Tuple2<String,String>, DomainHistory>() {

			@Override
			public DomainHistory call(Tuple2<String, String> tuple) throws Exception {
				DomainHistory dh = new DomainHistory();
				try{
					String domain = tuple._1;
					if(StringUtils.isNotBlank(domain)){
						String value = tuple._2;
						String[] valueStr = value.split("@@");
						List<String> list = new ArrayList<String>();
						List<Long> timeList = new ArrayList<Long>();
						StringBuffer ipBuffer = new StringBuffer();
						List<String> effectiveList = new ArrayList<String>();
						List<String> aliveList = new ArrayList<String>();
						for(String str : valueStr){
							String[] strSplit = str.split("@");
							String ip = strSplit[0];
							String timestamps = strSplit[1];
							String isEffective = strSplit[2];
							String isAlive = strSplit[3];
							if(!list.contains(ip)){//IP去重
								list.add(ip);
							}
							timeList.add(Long.parseLong(timestamps));
							effectiveList.add(isEffective);
							aliveList.add(isAlive);
						}
						for(String ip : list){
							ipBuffer.append(ip).append("|");
						}
						long isAlive = getFilterResult(aliveList);
						long isEffective = getFilterResult(effectiveList);
						dh.setDomain(domain);
						dh.setIp(getFilter(ipBuffer));
						dh.setTimestamps(Collections.min(timeList));//取发现时间最短的
						dh.setIsAlive(isAlive);
						dh.setIsEffective(isEffective);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return dh;
			}
		});
		
		DataFrame frame = sqlContext.createDataFrame(domainRDD, DomainHistory.class);
		frame.registerTempTable("domain_history_filter_temp");
		
		DataFrame domainFilterFrame = sqlContext.sql("select domain,ip,timestamps,isEffective,isAlive from domain_history_filter_temp where domain is not null");
		
		return domainFilterFrame;
	}

	@SuppressWarnings("serial")
	private static JavaRDD<DomainIp> getDayDomainRDD(SQLContext sqlContext,final Broadcast<IPDatabase> ipBroadcast, final List<String> surfilterIpList,String hiveDB, String date) {
		String sql = "select domain,ip,timestamps from "+hiveDB+".day_domain t where t.ds = '"+date+"'";
		DataFrame dayDomainFrame = sqlContext.sql(sql);
		
		JavaRDD<DomainIp> dayDomainRDD = dayDomainFrame.toJavaRDD().map(new Function<Row,DomainIp>() {

			@Override
			public DomainIp call(Row row) throws Exception {
				DomainIp domainIp = new DomainIp();
				try{
					String ip = row.getAs("ip");
					StringBuffer ipBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(ip)){
						String[] ipSplit = ip.split("\\|");
						for(String ipStr: ipSplit){
							StructIP structIp = ipBroadcast.value().get(ipStr);
							//过滤外省IP
							if(structIp != null){//为null则为外省
								//过滤114IP
								if(!surfilterIpList.contains(ipStr)){
									ipBuffer.append(ipStr).append("|");
								}
							}
						}
					}
					if(ipBuffer.length() > 0){
						String domain = row.getAs("domain");
						long timestamps = Long.parseLong(row.getAs("timestamps").toString());
						domainIp.setDomain(domain);
						domainIp.setIp(getFilter(ipBuffer));
						domainIp.setTimestamps(timestamps);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return domainIp;
			}
		});
		
		return dayDomainRDD;
	}
	
	/**
	 * 处理逻辑：相同的域名进行合并，同时合并IP并去重
	 * @param sqlContext
	 * @param dayDomainRDD
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getMergeDayDomainFrame(SQLContext sqlContext,JavaRDD<DomainIp> dayDomainRDD) {
		JavaRDD<DomainIp> domainRDD = dayDomainRDD.mapToPair(new PairFunction<DomainIp,String,String>() {

			@Override
			public Tuple2<String,String> call(DomainIp domainIp){
				try{
					String domain = domainIp.getDomain();
					String ip = domainIp.getIp();
					if(StringUtils.isNotBlank(domain)){
						return new Tuple2<String,String>(domain,ip+"@"+domainIp.getTimestamps());
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","");
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2) throws Exception {
				return v1 + "@@" + v2;
			}
		}).map(new Function<Tuple2<String,String>, DomainIp>() {

			@Override
			public DomainIp call(Tuple2<String, String> tuple) throws Exception {
				DomainIp domainIp = new DomainIp();
				try{
					String domain = tuple._1;
					if(StringUtils.isNotBlank(domain)){
						String value = tuple._2;
						String[] valueStr = value.split("@@");
						List<String> list = new ArrayList<String>();
						List<Long> timeList = new ArrayList<Long>();
						StringBuffer ipBuffer = new StringBuffer();
						for(String ipTime : valueStr){
							String[] ipTimeStr = ipTime.split("@");
							String ip = ipTimeStr[0];
							String timestamps = ipTimeStr[1];
							if(!list.contains(ip)){//IP去重
								list.add(ip);
							}
							timeList.add(Long.parseLong(timestamps));
						}
						for(String ip : list){
							ipBuffer.append(ip).append("|");
						}
						domainIp.setDomain(domain);
						domainIp.setIp(getFilter(ipBuffer));
						domainIp.setTimestamps(Collections.min(timeList));//取发现时间最短的
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return domainIp;
			}
		});
		
		DataFrame frame = sqlContext.createDataFrame(domainRDD, DomainIp.class);
		frame.registerTempTable("day_domain_filter_temp");
		
		DataFrame domainFilterFrame = sqlContext.sql("select domain,ip,timestamps from day_domain_filter_temp where domain is not null");
		
		return domainFilterFrame;
	}

	private static DataFrame getDomainHistroyV3Frame(SQLContext sqlContext,String date,String hiveDB) {
		DataFrame domainHisV3Frame = null;
		try{
			StringBuffer sql = new StringBuffer();
			sql.append("select history.domain      his_domain,");
			sql.append(" 	   history.ip          his_ip,");
			sql.append("	   history.timestamps,");
			sql.append("       history.isEffective,");
			sql.append("       history.isAlive,");
			sql.append("       yd.domain           yd_domain,");
			sql.append("       yd.ip               yd_ip ");
			sql.append("from domain_history_v2_temp history ");
			sql.append("full join day_domain_temp yd ");
			sql.append("on history.domain = yd.domain");
			
			domainHisV3Frame = sqlContext.sql(sql.toString());
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return domainHisV3Frame;
	}

	/**
	 * 处理逻辑: 最终改变的是IP
	 *         1、域名同时存在历史域名和域名IP关系中，若域名IP关系中IP为空，则删除该历史域名记录；否则更新历史域名中的IP
	 *         2、域名只存在历史域名中，则历史域名中的记录保持不变
	 * @param sqlContext
	 * @param date
	 * @param hiveDB
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getDomainHistoryV1Frame(SQLContext sqlContext,String date,String hiveDB) {
		DataFrame domainHistoryV1Frame = null;
		StringBuffer sql = new StringBuffer();
		sql.append("select dh.domain,dh.ip as dh_ip,dh.timestamps, dh.isEffective, dh.isAlive,dit.domain as dit_domain,dit.ip as dit_ip");
		sql.append("  from domain_history_temp dh left join domain_ip_temp dit");
		sql.append("   on dit.domain = dh.domain");
		
		domainHistoryV1Frame = sqlContext.sql(sql.toString());
		
		JavaRDD<DomainHistory> rdd = domainHistoryV1Frame.toJavaRDD().map(new Function<Row,DomainHistory>() {

			@Override
			public DomainHistory call(Row row){
				DomainHistory dh = new DomainHistory();
				try{
					String domain = row.getAs("domain");
					String dh_ip = row.getAs("dh_ip");
					String dit_domain = row.getAs("dit_domain");
					String dit_ip = row.getAs("dit_ip");
					long timestamps = Long.parseLong(row.getAs("timestamps").toString());
					long isEffective = Long.parseLong(row.getAs("isEffective").toString());
					long isAlive = Long.parseLong(row.getAs("isAlive").toString());
					
					if(StringUtils.isNotBlank(dit_domain)){
						if(StringUtils.isBlank(dit_ip)){
							//删除历史域名中该条记录
							return dh;
						}else{
							dh.setIp(dit_ip);
						}
					}else{
						dh.setIp(dh_ip);
					}
					
					dh.setDomain(domain);
					dh.setIsAlive(isAlive);
					dh.setTimestamps(timestamps);
					dh.setIsEffective(isEffective);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return dh;
			}
		});
			
		DataFrame df = sqlContext.createDataFrame(rdd, DomainHistory.class);
		df.registerTempTable("domain_history_filter_v1_temp");
		DataFrame domainHistoryFrame = sqlContext.sql("select * from domain_history_filter_v1_temp where domain is not null");
		return domainHistoryFrame;
	}

	public static void insertBr2005(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD){
		String sql = "";
		boolean flag = true;
		Connection conn = null;
		try{
			conn = JdbcUtil.getConnection();
			
			long num = getUserTablesCount(conn,"BR2005BAK");
			if(num > 0){
				//1.删除备份表
				sql = "drop table br2005bak";
				flag = execSql(conn,sql);
			}
			if(flag){
				 //2.创建备份表
				sql = " create table br2005bak as select * from br2005";
				flag = execSql(conn,sql);
				if(flag){
					//3.将表br2005字段is_alive、join_count、is_effective全部更新为0
					sql = "update br2005 set join_count = 0 , is_alive = 0 , is_effective = 0";
					execSql(conn,sql);
					
					//4.删除br2005_mg表数据
					sql = "truncate table br2005_mg";
					execSql(conn,sql);
					
					//5.将结果集插入br2005_mg表中
					insertBr2005Mg(webSiteIpSurvivalRDD,conn);
					
					//6.对表br2005数据进行操作，若数据集中域名在表中不存在则新增，否则更新
					insertAndUpdateTable(conn);
					
					//7.处理br2005中多余顶级域名数据
					sql = "delete from br2005 where join_count <= 0";
					execSql(conn,sql);
				}
			}
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(conn);
		}
	}
	
	public static long getUserTablesCount(Connection conn,String table) {
		long num = 0;
		String sql = "select count(1) as count from user_tables where table_name = ?";
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			ps = conn.prepareStatement(sql);
			ps.setString(1,table);
			rs = ps.executeQuery();
			if (rs.next()) {
				num = rs.getLong(1);
			}
		} catch (SQLException e) {
			num = 0;
			log.error("jdbc batch commit sql:" + sql + " error", e);
		}finally{
			JdbcUtil.close(rs);
			JdbcUtil.close(ps);
		}
		return num;
	}
	
	public static void insertBr2006(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD,Broadcast<IPDatabase> ipBroadcast){
		Connection conn = null;
		try{
			String sql = "";
			boolean flag = true;
			conn = JdbcUtil.getConnection();
			long num = getUserTablesCount(conn,"BR2006BAK");
			if(num > 0){
				//删除备份表
				sql = "drop table br2006bak";
				flag = execSql(conn,sql);
			}
			if(flag){
				 //创建备份表
				sql = " create table br2006bak as select * from br2006";
				flag = execSql(conn,sql);
				if(flag){
					sql = "truncate table br2006";
					execSql(conn,sql);
					
					operBr2006(webSiteIpSurvivalRDD,ipBroadcast,conn);
				}
			}
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(conn);
		}
	}
	
	private static void operBr2006_bak(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD,final Broadcast<IPDatabase> ipBroadcast) {
		
		webSiteIpSurvivalRDD.foreachPartition(new VoidFunction<Iterator<WebSiteIpSurvival>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<WebSiteIpSurvival> it){
				Connection conn = null;
				PreparedStatement ps = null;
				try {
					conn = JdbcUtil.getConnection();
					
					String insertSql = "insert into br2006(id,top_domain_id,ip_addr,ip_addr_value,idc_isp_id,house_id,province,city,county,top_domain,house_code,operators_id,is_alive,is_cdn,isp_id) "
							+ "values(BR2006_SE.nextval,1,?,?,?,?,?,?,?,?,?,?,?,?,?)";

					conn.setAutoCommit(false);
					ps = conn.prepareStatement(insertSql);
					int n = 0;
					
					while (it.hasNext()) {
						WebSiteIpSurvival website = it.next();
						
						String ips = website.getIp();
						String isAlives = website.getIsAlive();
						String isCdns= website.getIsCdn();
						
						String[] ipStr = ips.split("\\|");
						String[] aliveStr = isAlives.split("\\|");
						String[] cdnStr = isCdns.split("\\|");
						for(int i = 0 ; i < ipStr.length ; i++){
							String ip = ipStr[i];
							String isAlive = aliveStr[i];
							String isCdn = cdnStr[i];
							
							StructIP structIp = ipBroadcast.value().get(ip);
							if(structIp != null){
								ps.setString(1, ip);//ip_addr
								ps.setString(2,String.valueOf(DNSParser.ip2Long(ip)));//ip_addr_value
								ps.setString(3, structIp.get(DnsStatConsts.IPExField.ISP_JYXKZBH));//idc_isp_id
								ps.setLong(4,222);//house_id
								
								String province = structIp.get(DnsStatConsts.IPExField.PROVINCE);
								if(StringUtils.isNotBlank(province) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, province)){
									ps.setString(5,null);
								}else{
									ps.setString(5,province);
								}
								
								String city = structIp.get(DnsStatConsts.IPExField.CITY);
								if(StringUtils.isNotBlank(city) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, city)){
									ps.setString(6,null);
								}else{
									ps.setString(6,city);
								}
							
								String county = structIp.get(DnsStatConsts.IPExField.COUNTY);
								if(StringUtils.isNotBlank(county) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, county)){
									ps.setString(7,null);
								}else{
									ps.setString(7,county);
								}
								ps.setString(8, website.getDomain());
								
								String house_code = structIp.get(DnsStatConsts.IPExField.HOUSE_CODE);
								if(StringUtils.isNotBlank(house_code) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, house_code)){
									ps.setString(9,null);
								}else{
									ps.setString(9,house_code);
								}
								
								ps.setString(10, structIp.get(DnsStatConsts.IPExField.OPERATORS_ID));
								ps.setLong(11,parseLong(isAlive));
								Set<String> tm0105Set = getTm0105IpInfo();
								//先用DIP查询表tm0105 如果有数据，该值设置为1
								if(tm0105Set.contains(ip)){
									ps.setLong(12,1);
								}else{
									ps.setLong(12,parseLong(isCdn));
								}
								String isp_id = structIp.get(DnsStatConsts.IPExField.ISP_ID);
								if(StringUtils.isNotBlank(isp_id) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, isp_id)){
									ps.setString(13,null);
								}else{
									ps.setString(13,isp_id);
								}
								ps.addBatch();
								
								if (++n % 5000 == 0) {
									ps.executeBatch();
									conn.commit();
								}
							}
						}
					}
					ps.executeBatch();
					conn.commit();
				} catch (Exception e) {
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				} finally {
					JdbcUtil.close(ps,conn);
				}
			}
		});
	}

	
	private static void operBr2006(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD,final Broadcast<IPDatabase> ipBroadcast,Connection conn) {
		List<WebSiteIpSurvival> list = webSiteIpSurvivalRDD.collect();
		Set<String> tm0105Set = getTm0105IpInfo();
		IPDatabase ipDatabase = ipBroadcast.getValue();
		
		if(list != null && list.size() > 0){
			//Connection conn = null;
			PreparedStatement ps = null;
			try {
				//conn = JdbcUtil.getConnection();
				
				String insertSql = "insert into br2006(id,top_domain_id,ip_addr,ip_addr_value,idc_isp_id,house_id,province,city,county,top_domain,house_code,operators_id,is_alive,is_cdn,isp_id) "
						+ "values(BR2006_SE.nextval,1,?,?,?,?,?,?,?,?,?,?,?,?,?)";

				conn.setAutoCommit(false);
				ps = conn.prepareStatement(insertSql);
				int n = 0;
				for(WebSiteIpSurvival website : list){

					String ips = website.getIp();
					String isAlives = website.getIsAlive();
					String isCdns= website.getIsCdn();
					
					String[] ipStr = ips.split("\\|");
					String[] aliveStr = isAlives.split("\\|");
					String[] cdnStr = isCdns.split("\\|");
					for(int i = 0 ; i < ipStr.length ; i++){
						String ip = ipStr[i];
						String isAlive = aliveStr[i];
						String isCdn = cdnStr[i];
						
						StructIP structIp = ipDatabase.get(ip);
						if(structIp != null){
							ps.setString(1, ip);//ip_addr
							ps.setString(2,String.valueOf(DNSParser.ip2Long(ip)));//ip_addr_value
							ps.setString(3, structIp.get(DnsStatConsts.IPExField.ISP_JYXKZBH));//idc_isp_id
							ps.setLong(4,222);//house_id
							
							String province = structIp.get(DnsStatConsts.IPExField.PROVINCE);
							if(StringUtils.isNotBlank(province) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, province)){
								ps.setString(5,null);
							}else{
								ps.setString(5,province);
							}
							
							String city = structIp.get(DnsStatConsts.IPExField.CITY);
							if(StringUtils.isNotBlank(city) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, city)){
								ps.setString(6,null);
							}else{
								ps.setString(6,city);
							}
						
							String county = structIp.get(DnsStatConsts.IPExField.COUNTY);
							if(StringUtils.isNotBlank(county) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, county)){
								ps.setString(7,null);
							}else{
								ps.setString(7,county);
							}
							ps.setString(8, website.getDomain());
							
							String house_code = structIp.get(DnsStatConsts.IPExField.HOUSE_CODE);
							if(StringUtils.isNotBlank(house_code) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, house_code)){
								ps.setString(9,null);
							}else{
								ps.setString(9,house_code);
							}
							
							ps.setString(10, structIp.get(DnsStatConsts.IPExField.OPERATORS_ID));
							ps.setLong(11,parseLong(isAlive));
							
							//先用DIP查询表tm0105 如果有数据，该值设置为1
							if(tm0105Set.contains(ip)){
								ps.setLong(12,1);
							}else{
								ps.setLong(12,parseLong(isCdn));
							}
							String isp_id = structIp.get(DnsStatConsts.IPExField.ISP_ID);
							if(StringUtils.isNotBlank(isp_id) && StringUtils.equalsIgnoreCase(DnsStatConsts.NA_STR, isp_id)){
								ps.setString(13,null);
							}else{
								ps.setString(13,isp_id);
							}
							ps.addBatch();
							
							if (++n % 5000 == 0) {
								ps.executeBatch();
								conn.commit();
							}
						}
					}
				}
				ps.executeBatch();
				conn.commit();
			} catch (Exception e) {
				if(log.isErrorEnabled()){
					log.error(e.getMessage(),e);
				}
			} finally {
				//JdbcUtil.close(ps,conn);
				JdbcUtil.close(ps);
			}
		}
	}
	
	public static Set<String> getTm0105IpInfo(){
		
		StringBuffer sql = new StringBuffer();
		sql.append("select ip_addr");
		sql.append("  from (select a.*, rownum rn");
		sql.append("          from (select ip_addr from tm0105 ) a");
		sql.append("         where rownum <= ?) aa");
		sql.append("  where rn > ?");
		
		int min = 0;
		int max = 10000;
		int pageSize = 10000;
		int totalCount = 0;
		Set<String> tm0105Set = new HashSet<String>();
		Connection conn = null;
		PreparedStatement ps = null;
		ResultSet rs = null;
		try{
			conn = JdbcUtil.getConnection();
			ps =  conn.prepareStatement(sql.toString());
			do{
				ps.setInt(1,max);
				ps.setInt(2,min);
				rs = ps.executeQuery();
				List<String> list = new ArrayList<String>();
				while(rs.next()){
					String ip_addr = rs.getString("ip_addr");
					list.add(ip_addr);
				}
				tm0105Set.addAll(list);
				totalCount = list.size();
				min = max;
				max = max + pageSize;
			}while(pageSize == totalCount);
		}catch(SQLException e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(rs, ps, conn);
		}
		return tm0105Set;
	}
	
	private static void insertBr2005Mg(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD,Connection conn) {
		List<WebSiteIpSurvival> list = webSiteIpSurvivalRDD.collect();
		if(list != null && list.size() > 0){

			//Connection conn = null;
			PreparedStatement ps = null;
			try {
				//conn = JdbcUtil.getConnection();
				
				String insertSql = "insert into br2005_mg(id,top_domain,is_alive,join_count,find_time,sf_cdn,is_effective) "
						+ "values(BR2005_MG_SE.nextval,?,?,?,?,?,?)";

				conn.setAutoCommit(false);
				ps = conn.prepareStatement(insertSql);
				int n = 0;
				for(WebSiteIpSurvival website : list){
					String isAlive = website.getIsAlive();
					List<String> aliveList = Arrays.asList(isAlive.split("\\|"));
					int alive = 1;
					if(aliveList.contains("1")){
						alive = 1;
					}else if(aliveList.contains("0")){
						alive = 0;
					}else{
						alive = -1;
					}
					long time =	website.getTimestamps();
					long joinCount = website.getJoin_count();
					String isCdn = website.getIsCdn();
					List<String> cdnList = Arrays.asList(isCdn.split("\\|"));
					int cdn = 1;
					if(cdnList.contains("1")){
						cdn = 1;
					}else if(cdnList.contains("0")){
						cdn = 0;
					}else{
						cdn = -1;
					}
					ps.setString(1,website.getDomain());
					ps.setInt(2,alive);
					ps.setLong(3,joinCount);
					ps.setDate(4,new java.sql.Date(time * 1000));
					ps.setInt(5,cdn);
					ps.setLong(6,website.getIsEffective());
					ps.addBatch();
					
					if (++n % 5000 == 0) {
						ps.executeBatch();
						conn.commit();
					}
				}
				ps.executeBatch();
				conn.commit();
			} catch (Exception e) {
				if(log.isErrorEnabled()){
					log.error(e.getMessage(),e);
				}
			} finally {
				//JdbcUtil.close(ps,conn);
				JdbcUtil.close(ps);
			}
		
		}
	}
	
	private static void insertBr2005Mg_bak(JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD) {
		
		webSiteIpSurvivalRDD.foreachPartition(new VoidFunction<Iterator<WebSiteIpSurvival>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<WebSiteIpSurvival> it) throws Exception {
				Connection conn = null;
				PreparedStatement ps = null;
				try {
					conn = JdbcUtil.getConnection();
					
					String insertSql = "insert into br2005_mg(id,top_domain,is_alive,join_count,find_time,sf_cdn,is_effective) "
							+ "values(BR2005_MG_SE.nextval,?,?,?,?,?,?)";

					conn.setAutoCommit(false);
					ps = conn.prepareStatement(insertSql);
					int n = 0;
					
					while (it.hasNext()) {
						WebSiteIpSurvival website = it.next();
						
						String isAlive = website.getIsAlive();
						List<String> aliveList = Arrays.asList(isAlive.split("\\|"));
						int alive = 1;
						if(aliveList.contains("1")){
							alive = 1;
						}else if(aliveList.contains("0")){
							alive = 0;
						}else{
							alive = -1;
						}
						long time =	website.getTimestamps();
						long joinCount = website.getJoin_count();
						String isCdn = website.getIsCdn();
						List<String> cdnList = Arrays.asList(isCdn.split("\\|"));
						int cdn = 1;
						if(cdnList.contains("1")){
							cdn = 1;
						}else if(cdnList.contains("0")){
							cdn = 0;
						}else{
							cdn = -1;
						}
						ps.setString(1,website.getDomain());
						ps.setInt(2,alive);
						ps.setLong(3,joinCount);
						ps.setDate(4,new java.sql.Date(time * 1000));
						ps.setInt(5,cdn);
						ps.setLong(6,website.getIsEffective());
						ps.addBatch();
						
						if (++n % 5000 == 0) {
							ps.executeBatch();
							conn.commit();
						}
					}
					ps.executeBatch();
					conn.commit();
				} catch (Exception e) {
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				} finally {
					JdbcUtil.close(ps,conn);
				}
			}
		});
	}

	private static boolean execSql(Connection conn,String sql){
    	boolean flag = true;
    	PreparedStatement ps = null;
    	try{
    		conn.setAutoCommit(false);
    		ps = conn.prepareStatement(sql);
          	ps.executeQuery();
            conn.commit();
          	flag = true;
          }
          catch(SQLException e){
         	 flag = false;
         	 log.error("jdbc batch commit sql:"+sql+" error" , e);
              try {
                  conn.rollback();
              } catch (SQLException e1) {
                  log.error("jdbc rollback error" , e1);
              }
          }finally{
        	  JdbcUtil.close(ps);
          }
    	return flag ;
    }
	
	public static void updateTable(Connection conn){
		PreparedStatement ps = null;
		try {
			String sql = "update br2005 set join_count = ? , is_alive = ? , is_effective = ?";
			ps = conn.prepareStatement(sql.toString());
			ps.setObject(1,0);
			ps.setObject(2,0);
			ps.setObject(3,0);
			ps.executeUpdate();
			conn.commit();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps);
		}
	}
	
	public static void insertAndUpdateTable(Connection conn){
		
		StringBuffer sql = new StringBuffer();
		sql.append("merge into br2005 a ");
		sql.append("using br2005_mg b on (a.top_domain=b.top_domain) ");
		sql.append("when not matched then ");
		sql.append("insert(id,top_domain,is_alive,join_count,find_time,is_effective,industry_type,parent_industry_type,sf_cdn) ");
		sql.append("values(BR2005_SE.nextval,b.top_domain,b.is_alive,b.join_count,b.find_time,b.is_effective,220,216,b.sf_cdn)");
		sql.append("when matched then ");
		sql.append("update set a.is_alive = b.is_alive,a.join_count = b.join_count,a.is_effective = b.is_effective");

		PreparedStatement ps = null;
		try {
			conn.setAutoCommit(false);
			ps = conn.prepareStatement(sql.toString());
			ps.executeUpdate();
			conn.commit();
			conn.setAutoCommit(true);
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps);
		}
	}
	
	public static void dropTable(Connection conn,String table){
		PreparedStatement ps = null;
		String delSql = "drop table "+table;
		try {
			ps = conn.prepareStatement(delSql);
			ps.executeUpdate();
			conn.commit();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps);
		}
	}
	
	public static void deleteTable(Connection conn,String table){
		PreparedStatement ps = null;
		String delSql = "truncate table "+table;
		try {
			ps = conn.prepareStatement(delSql);
			ps.executeUpdate();
			conn.commit();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps);
		}
	}
	
	@SuppressWarnings("serial")
	private static void insertTopDomainToEs(String esNode,final Broadcast<IPDatabase> ipBroadcast, 
			final Broadcast<DomainDatabase> domainBroadcast,JavaRDD<WebSiteIpSurvival> webSiteIpSurvivalRDD,String date,SysProps sysProps){
		
		JavaRDD<TopDomainIndex> topDomainRdd = webSiteIpSurvivalRDD.map(new Function<WebSiteIpSurvival,TopDomainIndex>() {

			@Override
			public TopDomainIndex call(WebSiteIpSurvival site){
				TopDomainIndex index = new TopDomainIndex();
				try{
					String domain = site.getDomain();
					
					index.setId(0);
					index.setTopDomain(domain);
					
					StructDomain structDomain = domainBroadcast.value().get(domain);
					//structDomain为null时，ES表所对应的字段值为null，不需要额外的设置
					if(structDomain != null){
						
						String siteName = structDomain.get(DnsStatConsts.DomainExField.SITE_NAME);
						if(StringUtils.isNotBlank(siteName)){
							index.setSiteName(siteName);
						}else{
							index.setSiteName("");
						}
						
						String licenseKey = structDomain.get(DnsStatConsts.DomainExField.LICENSE_KEY);
						if(StringUtils.isNotBlank(licenseKey)){
							index.setLicenseKey(licenseKey);
						}else{
							index.setLicenseKey("");
						}
						String recordState = structDomain.get(DnsStatConsts.DomainExField.RECORD_STATE);
						if(StringUtils.isNotBlank(recordState)){
							index.setRecordState(parseLong(recordState));
						}
						
						String isBlack = structDomain.get(DnsStatConsts.DomainExField.IS_BLACK);
						if(StringUtils.isNotBlank(isBlack)){
							index.setIsBlack(parseLong(isBlack));
						}
						
						String isAlive = site.getIsAlive();
						List<String> aliveList = Arrays.asList(isAlive.split("\\|"));
						long alive = getFilterResult(aliveList);
						
						index.setIsAlive(alive);
						
						index.setRegistrarName(structDomain.get(DnsStatConsts.DomainExField.REGISTRAR_NAME));
						index.setCnDomainName(structDomain.get(DnsStatConsts.DomainExField.CN_DOMAIN_NAME));
						index.setIndustryType(structDomain.get(DnsStatConsts.DomainExField.PARENT_INDUSTRY_TYPE));
						index.setIndustryTypeName("");//有数据为空字符;无数据时为null
						
						long timestamps = site.getTimestamps();
						Date date = new Date(timestamps*1000);
						SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
						String findTime = sdf.format(date);
						index.setFindTime(findTime);
						
					}
					
					StringBuffer ispIdBuffer = new StringBuffer();//接入商Id集合
					StringBuffer ispNameBuffer = new StringBuffer();//接入商名称集合
					StringBuffer idcIspIdBuffer = new StringBuffer();//接入商经营许可证号集合
					StringBuffer houseCodeBuffer = new StringBuffer();//机房编码集合
					StringBuilder provinceBuffer = new StringBuilder();//省编码集合
					StringBuffer cityBuffer = new StringBuffer();//市编码集合
					StringBuffer countyBuffer = new StringBuffer();//县编码集合
					StringBuffer ipBuffer = new StringBuffer();//IP集合
					
					String ip = site.getIp();
					
					if(StringUtils.isNotBlank(ip)){
						String[] ipSplit = ip.split("\\|");
						for(String ipStr : ipSplit){
							if(StringUtils.isNotBlank(ipStr)){
								StructIP structIp = ipBroadcast.value().get(ipStr);
								if(structIp == null){
									continue;
								}
								ipBuffer.append(ipStr).append(",");//ip集合
								ispIdBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_ID)).append(",");//接入商Id集合
								ispNameBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_NAME)).append(",");//接入商名称集合
								idcIspIdBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_JYXKZBH)).append(",");//接入商经营许可证号集合
								houseCodeBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.HOUSE_CODE)).append(",");//机房编码集合
								provinceBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.PROVINCE)).append(",");//省编码集合
								cityBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.CITY)).append(",");//市编码集合
								countyBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.COUNTY)).append(",");//县编码集合
							}
						}
					}
					String splitStr = ",";
					String[] ips = ipBuffer.toString().split(splitStr);
					index.setIps(ips);
					index.setJrsIds(transformLongArray(ispIdBuffer.toString(),splitStr,ips.length));
					index.setJrsMcs(transformStringArray(ispNameBuffer.toString(),splitStr,ips.length));
					index.setIdcIspIds(transformStringArray(idcIspIdBuffer.toString(),splitStr,ips.length));
					index.setHouseCodes(transformStringArray(houseCodeBuffer.toString(),splitStr,ips.length));
					index.setShengIds(transformLongArray(provinceBuffer.toString(),splitStr,ips.length));
					index.setShiIds(transformLongArray(cityBuffer.toString(),splitStr,ips.length));
					index.setXianIds(transformLongArray(countyBuffer.toString(),splitStr,ips.length));
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return index;
			}

		});
		
		try{
			EsClientPool pool = null;
			EsClient esClient = null;
			int clientNum = sysProps.getInteger("es.clientNum", 1);
			
			//节点间交互的tcp端口
			String tcpPort = sysProps.getValue("transport.tcp.port");
			String clusterName = sysProps.getValue("es.clusterName");
			pool = new EsClientPool(clusterName,esNode, tcpPort, clientNum);
			esClient = pool.getEsClient();
			
			String dateStr = DateUtils.formatDateToES(DateUtils.stringToDate(date));
			String indexName = "dns_" + dateStr +"_provincetld";
			String type = "ptld001";
			
			IndexClass indexClass = new IndexClass(indexName,type);
			EsEntity<TopDomainIndex> esEntity = new EsTopDomainVisits();
			EsMappingService es = new EsMappingServiceImpl(indexClass, pool, esEntity);
			
			// 先删除索引,避免数据重复录入同一索引
			if (esClient.isExists(indexName)) {
				esClient.deleteIndex(indexName);
				log.info("---------删除已存在的索引----------" + indexName);
			}
			String number_of_shards = sysProps.getValue("es.number_of_shards", "3");
			String number_of_replicas = sysProps.getValue("es.number_of_replicas", "0");
			es.init(number_of_shards,number_of_replicas);
			
			Map<String,String> map = new HashMap<String,String>();
			map.put("es.mapping.id","topDomain");
			JavaEsSpark.saveToEs(topDomainRdd, "/"+indexName+"/"+type,map);
			
			//清除ES数据
			int keeyDate = sysProps.getInteger("esindex.keep.day",30); 
			deleteIndex(pool,"provincetld",keeyDate);
			
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	/**
	 * 处理逻辑:1.网站IP存活关系V1在天域名中找到对应关系，合并对应的IP，并将其对应的存活状态和isCdn设置为-1
	 *        2.网站IP存活关系V1在天域名中没有找到对应关系,该条记录保持不变
	 *        3.天域名中在网站IP存活关系V1中没有找到对应关系，将其对应的存活状态和isCdn设置为-1
	 * @param sqlContext
	 * @return
	 */
	@SuppressWarnings("serial")
	private static JavaRDD<WebSiteIpSurvival> getDayHisSurCdnMergeRDD(SQLContext sqlContext) {
		
		StringBuffer sql = new StringBuffer();
		sql.append("select hscmt.domain     as hscmtDomain,");
		sql.append("       hscmt.ip         as hscmtIp,");
		sql.append("       hscmt.isAlive,");
		sql.append("       hscmt.isCdn,");
		sql.append("       hscmt.join_count as hscmtJoinCount,");
		sql.append("       hscmt.timestamps as hscmtTimestamps,");
		sql.append("       hscmt.fullDomain as hscmtFullDomain,");
		sql.append("       hscmt.isEffective,");
		sql.append("       ddmt.domain      as ddmtDomain,");
		sql.append("       ddmt.fullDomain  as ddmtFullDomain,");
		sql.append("       ddmt.ip          as ddmtIp,");
		sql.append("       ddmt.join_count  as ddmtJoinCount,");
		sql.append("       ddmt.timestamps  as ddmtTimestamps");
		sql.append("  from his_sur_cdn_merge_temp hscmt");
		sql.append("  full join day_domain_merge_temp ddmt");
		sql.append("    on hscmt.domain = ddmt.domain");
		
		DataFrame hisSurCdnDayFrame = sqlContext.sql(sql.toString());
		
		/**
			+-----------+--------------------+-------+-----+--------------+---------------+---------------+------------+---------------+--------------------+-------------+--------------+
			|hscmtDomain|             hscmtIp|isAlive|isCdn|hscmtJoinCount|hscmtTimestamps|hscmtFullDomain|  ddmtDomain| ddmtFullDomain|              ddmtIp|ddmtJoinCount|ddmtTimestamps|
			+-----------+--------------------+-------+-----+--------------+---------------+---------------+------------+---------------+--------------------+-------------+--------------+
			|  18185.com|        27.16.24.255|     -1|   -1|             1|     1498755600| img7.18185.com|        null|           null|                null|         null|          null|
			|      2.com|          27.16.25.2|      0|    0|             1|     1498755600|          2.com|       2.com|www.2.com|2.com|27.16.0.110|27.16...|            2|    1498831200|
			|       null|                null|   null| null|          null|           null|           null|10dfghhh.com| 1.10dfghhh.com|           27.16.0.0|            1|    1498831200|
			+-----------+--------------------+-------+-----+--------------+---------------+---------------+------------+---------------+--------------------+-------------+--------------+
		 */
		
		JavaRDD<WebSiteIpSurvival> resultRDD = hisSurCdnDayFrame.toJavaRDD().map(new Function<Row, WebSiteIpSurvival>() {

			@Override
			public WebSiteIpSurvival call(Row row){
				WebSiteIpSurvival site = new WebSiteIpSurvival();
				try{
					String hscmtDomain = row.getAs("hscmtDomain");
					String ddmtDomain = row.getAs("ddmtDomain");
					
					if(StringUtils.isNotBlank(hscmtDomain) && StringUtils.isNotBlank(ddmtDomain)){
						String hscmtFullDomain = row.getAs("hscmtFullDomain");
						String ddmtFullDomain = row.getAs("ddmtFullDomain");
						String hscmtIp = row.getAs("hscmtIp");
						String isAlive = row.getAs("isAlive");
						String isCdn = row.getAs("isCdn");
						long hscmtTimestamps = Long.parseLong(row.getAs("hscmtTimestamps").toString());
						String ddmtIp = row.getAs("ddmtIp");
						long ddmtTimestamps = Long.parseLong(row.getAs("ddmtTimestamps").toString());
						long isEffective = Long.parseLong(row.getAs("isEffective").toString());
						
						long timestamps = (hscmtTimestamps - ddmtTimestamps >= 0) ? ddmtTimestamps : hscmtTimestamps; //获取时间短的
						 
						List<String> hscmtIpList = Arrays.asList(hscmtIp.split("\\|"));
						
						String[] ddmtIpStr = ddmtIp.split("\\|");
						StringBuffer hscmtIpBuffer = new StringBuffer();
						hscmtIpBuffer.append(hscmtIp).append("|");//之前所有操作已经过滤掉IP结尾的"|",故跟其他IP拼接需要加上
						
						StringBuffer aliveBuffer = new StringBuffer();
						aliveBuffer.append(isAlive).append("|");
						
						StringBuffer cdnBuffer = new StringBuffer();
						cdnBuffer.append(isCdn).append("|");
						
						for(String ipStr : ddmtIpStr){
							if(!hscmtIpList.contains(ipStr)){
								hscmtIpBuffer.append(ipStr).append("|");
								aliveBuffer.append("-1").append("|");
								cdnBuffer.append("-1").append("|");
							}
						}
						//计算接入数总量
						Set<String> joinCountSet = new HashSet<String>();
						//相同记录自动去重
						joinCountSet.addAll(Arrays.asList(hscmtFullDomain.split("\\|")));
						joinCountSet.addAll(Arrays.asList(ddmtFullDomain.split("\\|")));
						
						site.setDomain(hscmtDomain);
						site.setIp(getFilter(hscmtIpBuffer));
						site.setIsAlive(getFilter(aliveBuffer));
						site.setIsCdn(getFilter(cdnBuffer));
						site.setJoin_count(joinCountSet.size());
						site.setTimestamps(timestamps);
						site.setIsEffective(isEffective);
					}else if(StringUtils.isNotBlank(hscmtDomain) && StringUtils.isBlank(ddmtDomain)){
						String hscmtIp = row.getAs("hscmtIp");
						String isAlive = row.getAs("isAlive");
						String isCdn = row.getAs("isCdn");
						long join_count = Long.parseLong(row.getAs("hscmtJoinCount").toString());
						long timestamps = Long.parseLong(row.getAs("hscmtTimestamps").toString());
						long isEffective = Long.parseLong(row.getAs("isEffective").toString());
						
						site.setDomain(hscmtDomain);
						site.setIp(hscmtIp);
						site.setIsAlive(isAlive);
						site.setIsCdn(isCdn);
						site.setJoin_count(join_count);
						site.setTimestamps(timestamps);
						site.setIsEffective(isEffective);
					}else if(StringUtils.isBlank(hscmtDomain) && StringUtils.isNotBlank(ddmtDomain)){
						String ddmtIp = row.getAs("ddmtIp");
						long join_count = Long.parseLong(row.getAs("ddmtJoinCount").toString());
						long timestamps = Long.parseLong(row.getAs("ddmtTimestamps").toString());
						
						StringBuffer aliveBuffer = new StringBuffer();
						StringBuffer cdnBuffer = new StringBuffer();
						String[] ddmtIpStr = ddmtIp.split("\\|");
						for(int i = 0 ; i < ddmtIpStr.length ; i++){
							aliveBuffer.append("-1").append("|");
							cdnBuffer.append("-1").append("|");
						}
						
						site.setDomain(ddmtDomain);
						site.setIp(ddmtIp);
						site.setIsAlive(getFilter(aliveBuffer));
						site.setIsCdn(getFilter(cdnBuffer));
						site.setJoin_count(join_count);
						site.setTimestamps(timestamps);
						site.setIsEffective(1);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return site;
			}
		});
		
		return resultRDD;
	}

	@SuppressWarnings("serial")
	private static DataFrame getDayDomainMergeFrame(DataFrame dayDomainFrame,HiveContext hiveSql,SQLContext sqlContext,String date,final List<String> surfilterIpList,String hiveDB) {
		
		JavaPairRDD<String,String> dayDomainPiarRDD = dayDomainFrame.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() {

			@Override
			public Tuple2<String,String> call(Row row){
				try{
					String domain = row.getAs("domain");
					String ip = row.getAs("ip");
					String timestamps = row.getAs("timestamps").toString();
					String topDomain = URLUtil.getDomainName("http://" + domain);
					
					StringBuffer ipBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(topDomain)){
						String[] ipStr = ip.split("\\|");
						for(String str : ipStr){
							if(StringUtils.isNotBlank(str)){
								ipBuffer.append(str).append("|");
							}
						}
						//保留全量域名，用于和历史域名中的全量域名进行排重统计，计算接入数总量
						return new Tuple2<String,String>(topDomain,getFilter(ipBuffer)+"@"+timestamps+"@"+domain);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","");
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2){
				if(StringUtils.isNotBlank(v1) && StringUtils.isNotBlank(v2)){
					return v1 + "@@" + v2;
				}
				return "";
			}
		});

		//baidu.com = 49.82.0.0|49.83.0.0@1492790555@1.baidu.com@@49.82.0.0|49.83.0.0@1492790556@2.baidu.com
		JavaRDD<DomainIp> dayDomainRDD =  dayDomainPiarRDD.map(new Function<Tuple2<String,String>,DomainIp>() {

			@Override
			public DomainIp call(Tuple2<String, String> tuple) throws Exception {
				DomainIp domainIp = new DomainIp();
				try{
					if(tuple == null) return domainIp;
					String domain = tuple._1;
					String value = tuple._2;
					List<Long> timeList = new ArrayList<Long>();
					StringBuffer ipBuffer = new StringBuffer();
					StringBuffer fullDomainBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(value)){
						String[] valueStr = value.split("@@");
						for(String str : valueStr){
							String[] ipTimeDomain = str.split("@");
							String ips = ipTimeDomain[0];
							long time = Long.parseLong(ipTimeDomain[1]);
							String fullDomain = ipTimeDomain[2];
							String[] ipStr = ips.split("\\|");
							timeList.add(time);
							for(String ip : ipStr){
								//IP去重
								if(!ipBuffer.toString().contains(ip)){
									ipBuffer.append(ip).append("|");
								}
							}
							fullDomainBuffer.append(fullDomain).append("|");
						}
						domainIp.setDomain(domain);
						domainIp.setIp(getFilter(ipBuffer));
						domainIp.setTimestamps(Collections.min(timeList));//第一发现网站时间
						domainIp.setJoin_count(valueStr.length);//接入数,表示发现的该网站下的全量域名个数
						domainIp.setFullDomain(getFilter(fullDomainBuffer));
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return domainIp;
			}
		});
		
		DataFrame dayDomainResultFrame = sqlContext.createDataFrame(dayDomainRDD, DomainIp.class);
		
		return dayDomainResultFrame;
	}

	/**
	 * 处理逻辑:1.合并之后的历史域名与存活结果在CDN检测结果中没有找到对应关系，合并之后的历史域名与存活结果中的IP对应的isCdn全部设置为-1
	 *        2.若找到对应关系，则依次判断合并之后的历史域名与存活结果中的IP在CDN检测结果中否存在，若存在获取对应的isCdn中的值，否则设置为-1
	 * @param sqlContext
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getHistorySurCdnMergeFrame(SQLContext sqlContext) {
		
		StringBuffer sql = new StringBuffer();
		sql.append("select hsmt.domain,");
		sql.append("       hsmt.ip         as his_ip,");
		sql.append("       hsmt.isAlive,");
		sql.append("       hsmt.join_count,");
		sql.append("       hsmt.timestamps,");
		sql.append("       hsmt.fullDomain,");
		sql.append("       hsmt.isEffective,");
		sql.append("       dcmt.ip         as cdn_ip,");
		sql.append("       dcmt.isCdn");
		sql.append("  from history_survival_merge_temp hsmt");
		sql.append("  left join domain_cdn_merge_temp dcmt");
		sql.append("    on hsmt.domain = dcmt.domain");
		sql.append("   and dcmt.domain is not null");
		
		DataFrame hisSurCdnFrame = sqlContext.sql(sql.toString());
		
		/**
		 *  +---------+--------------------+-----------+----------+----------+--------------------+-----+
			|   domain|              his_ip|    isAlive|join_count|timestamps|              cdn_ip|isCdn|
			+---------+--------------------+-----------+----------+----------+--------------------+-----+
			|baidu.com|36.152.8.242|36.1...| -1|-1|0|1 |         3|1492790553|          49.85.0.0||   1||
			|weibo.com|36.152.9.16|49.82...|-1|1|-1|-1 |         2|1492790553|                null| null|
			+---------+--------------------+-----------+----------+----------+--------------------+-----+
		 */
		
		JavaRDD<WebSiteIpSurvival> hisSurCdnRDD = hisSurCdnFrame.toJavaRDD().map(new Function<Row, WebSiteIpSurvival>() {

			@Override
			public WebSiteIpSurvival call(Row row){
				WebSiteIpSurvival site = new WebSiteIpSurvival();
				try{
					String domain = row.getAs("domain");
					String fullDomain = row.getAs("fullDomain");
					String his_ip = row.getAs("his_ip");
					String isAlive = row.getAs("isAlive");
					long join_count = Long.parseLong(row.getAs("join_count").toString());
					long timestamps = Long.parseLong(row.getAs("timestamps").toString());
					String cdn_ip = row.getAs("cdn_ip");
					String isCdn = row.getAs("isCdn");
					long isEffective = Long.parseLong(row.getAs("isEffective").toString());
					
					String[] hisIpStr = his_ip.split("\\|");
					StringBuffer cdnBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(cdn_ip) && StringUtils.isNotBlank(isCdn)){
						String[] cdnIpStr = cdn_ip.split("\\|");
						String[] isCdnStr = isCdn.split("\\|");
						List<String> cdnIpList = Arrays.asList(cdnIpStr);
						Map<String,String> map = new HashMap<String,String>();
						for(int i = 0 ; i < cdnIpStr.length ; i++){
							map.put(cdnIpStr[i],isCdnStr[i]);
						}
						
						for(String hisIp : hisIpStr){
							if(cdnIpList.contains(hisIp)){
								//CDN中IP存在，则获取对应的isCdn的值
								cdnBuffer.append(map.get(hisIp)).append("|");
							}else{
								//不存在则将isCdn设置为-1
								cdnBuffer.append("-1").append("|");
							}
						}
					}else{
						for(int i = 0 ; i < hisIpStr.length ; i++){
							cdnBuffer.append("-1").append("|");
						}
					}
					site.setDomain(domain);
					site.setFullDomain(fullDomain);
					site.setIp(his_ip);
					site.setIsAlive(isAlive);
					site.setJoin_count(join_count);
					site.setTimestamps(timestamps);
					site.setIsCdn(getFilter(cdnBuffer));
					site.setIsEffective(isEffective);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return site;
			}
		});
		
		DataFrame historySurCdnFrame = sqlContext.createDataFrame(hisSurCdnRDD, WebSiteIpSurvival.class);

		return historySurCdnFrame;
	}

	@SuppressWarnings("serial")
	private static DataFrame getCdnInvalidMergeFrame(HiveContext hiveSql,SQLContext sqlContext, 
			final Broadcast<IPDatabase> ipBroadcast,final List<String> surfilterIpList,String date,String hiveDB) {
		
		DataFrame cdnInvalidFrame = hiveSql.sql("select domain,isCdn,ip from "+hiveDB+".cdn_invalid where ds = '"+date+"'");
		
		/**
		 *  +----------------+-----+---------+
			|          domain|isCdn|       ip|
			+----------------+-----+---------+
			|     a.apple.com|    1|49.85.0.0|
			|     b.apple.com|   -1|49.86.0.0|
			|     c.apple.com|   -1|		-|
			+----------------+-----+---------+
		 */
		
		JavaPairRDD<String,String> cdnPairRDD = cdnInvalidFrame.toJavaRDD().mapToPair(new PairFunction<Row,String, String>() {

			@Override
			public Tuple2<String,String> call(Row row) {
				try{
					String domain = row.getAs("domain");
					String isCdn = row.getAs("isCdn").toString();
					String ip = row.getAs("ip");
					
					String topDomain = URLUtil.getDomainName("http://" + domain);
					
					//过滤无效的域名和IP不存在
					if(StringUtils.isNotBlank(topDomain) && !StringUtils.equals(ip, "-")){
						//过滤外省IP
						StructIP structIp = ipBroadcast.value().get(ip);
						
						if(structIp != null){//为null则为外省
							//114 IP过滤
							if(surfilterIpList.contains(ip)){
								ip = "";
							}
						}else{
							ip = "";
						}
						if(StringUtils.isNotBlank(ip)){
							return new Tuple2<String,String>(topDomain,ip+"@"+isCdn);
						}
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","");
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2){
				if(StringUtils.isNotBlank(v1) && StringUtils.isNotBlank(v2)){
					return v1 + "@@" + v2;
				}
				return "";
			}
		});
		
		JavaRDD<CdnInvalid> cdnRDD = cdnPairRDD.map(new Function<Tuple2<String,String>, CdnInvalid>() {

			@Override
			public CdnInvalid call(Tuple2<String, String> tuple){
				CdnInvalid cdnInvalid = new CdnInvalid();
				try{
					if(tuple == null) return cdnInvalid;
					String domain = tuple._1;
					String value = tuple._2;
					StringBuffer ipBuffer = new StringBuffer();
					StringBuffer cdnBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(value)){
						String[] valueStr = value.split("@@");
						for(String str : valueStr){
							String[] ipCdn = str.split("@");
							String ip = ipCdn[0];
							String cdn = ipCdn[1];
							
							//IP去重
							if(!ipBuffer.toString().contains(ip)){
								ipBuffer.append(ip).append("|");
								cdnBuffer.append(cdn).append("|");
							}
						}
						cdnInvalid.setDomain(domain);
						cdnInvalid.setIp(getFilter(ipBuffer));
						cdnInvalid.setIsCdn(getFilter(cdnBuffer));
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return cdnInvalid;
			}
		});
		
		DataFrame cdnFrame = sqlContext.createDataFrame(cdnRDD, CdnInvalid.class);

		return cdnFrame;
	}
	/**
	 * 处理逻辑:1.若历史域名在存活结果中没有找到对应关系，则将历史域名的IP对应存活或者全部设置为-1，表示待检测
	 *        2.若找到对应关系，依次判断历史域名中的IP是在存活结果中是否存在，若存在则设置其对应的存活状态中的值，若不存在则设置为-1
	 * @param sqlContext
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getHistroySurvivalMergeFrame(SQLContext sqlContext) {
		
		StringBuffer sql = new StringBuffer();
		sql.append("select dhmt.domain,");
		sql.append("       dhmt.ip         as his_ip,");
		sql.append("       dhmt.timestamps,");
		sql.append("       dhmt.join_count,");
		sql.append("       dhmt.fullDomain,");
		sql.append("       dhmt.isEffective,");
		sql.append("       dsmt.ip         as sur_ip,");
		sql.append("       dsmt.isAlive");
		sql.append("  from domain_history_merge_temp dhmt");
		sql.append("  left join domain_survival_merge_temp dsmt");
		sql.append("    on dhmt.domain = dsmt.domain");
		
		DataFrame hisSurFrame = sqlContext.sql(sql.toString());
		
		/**
		 *  +---------+--------------------+----------+----------+--------------------+-------+
			|   domain|              his_ip|timestamps|join_count|              sur_ip|isAlive|
			+---------+--------------------+----------+----------+--------------------+-------+
			|apple.com|49.86.0.0|49.89.0.0||1492790553|         1|49.86.0.0|49.89.0.0||   1|1||
			+---------+--------------------+----------+----------+--------------------+-------+
		 */
		
		//若新历史域名V1.0中ip在存活结果中没有找到则将存在状态置为-1表示待检测,若找到则对应存活结果中存活状态值
		JavaRDD<WebSiteIpSurvival> webSiteRDD = hisSurFrame.toJavaRDD().map(new Function<Row,WebSiteIpSurvival>() {

			@Override
			public WebSiteIpSurvival call(Row row){
				WebSiteIpSurvival site = new WebSiteIpSurvival();
				try{
					String domain = row.getAs("domain");
					String fullDomain = row.getAs("fullDomain");
					String his_ip = row.getAs("his_ip");
					long timestamps = Long.parseLong(row.getAs("timestamps").toString());
					String sur_ip = row.getAs("sur_ip");
					String isAlive = row.getAs("isAlive");
					long join_count = Long.parseLong(row.getAs("join_count").toString());
					long isEffective = Long.parseLong(row.getAs("isEffective").toString());
					
					String[] hisIpStr = his_ip.split("\\|");
					StringBuffer ipBuffer = new StringBuffer();
					StringBuffer aliveBuffer = new StringBuffer();
					
					if(StringUtils.isNotBlank(sur_ip) && StringUtils.isNotBlank(isAlive)){
						String[] surIpStr = sur_ip.split("\\|");
						String[] isAliveStr = isAlive.split("\\|");
						List<String> surIpList = Arrays.asList(surIpStr);
						Map<String,String> map = new HashMap<String,String>();
						for(int i = 0 ; i < surIpStr.length ; i++){
							map.put(surIpStr[i],isAliveStr[i]);
						}
						for(String ip : hisIpStr){
							String alive = "";
							if(surIpList.contains(ip)){
								alive = map.get(ip);
							}else{
								alive = "-1";//没有找到对应IP表示待检测
							}
							ipBuffer.append(ip).append("|");
							aliveBuffer.append(alive).append("|");
						}
						site.setIp(getFilter(ipBuffer));
					}else{
						//域名没有对应关系，将其存活状态全部置为-1
						for(int i = 0 ; i < hisIpStr.length ; i++){
							aliveBuffer.append("-1").append("|");
						}
						site.setIp(his_ip);
					}
					site.setDomain(domain);
					site.setFullDomain(fullDomain);
					site.setIsAlive(getFilter(aliveBuffer));
					site.setTimestamps(timestamps);
					site.setJoin_count(join_count);
					site.setIsEffective(isEffective);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return site;
			}
		});
		
		DataFrame webSiteFrame = sqlContext.createDataFrame(webSiteRDD, WebSiteIpSurvival.class);
		
		return webSiteFrame;
	}
	
	@SuppressWarnings("serial")
	private static DataFrame getDomainSurvivalMergeFrame(SQLContext sqlContext,DataFrame domainSurvivalFrame) {
		
		JavaPairRDD<String,String> surPairRDD = domainSurvivalFrame.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() {

			@Override
			public Tuple2<String,String> call(Row row){
				try{
					String domain = row.getAs("domain");
					String ip = row.getAs("ip");
					String isAlive = row.getAs("isAlive").toString();
					
					String topDomain = URLUtil.getDomainName("http://" + domain);
					if(StringUtils.isNotBlank(topDomain)){
						return new Tuple2<String,String>(topDomain,ip+"@"+isAlive);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","");
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2){
				if(StringUtils.isNotBlank(v1) && StringUtils.isNotBlank(v2)){
					return v1 + "@@" + v2;
				}
				return "";
			}
		});
		
		JavaRDD<DomainSurvival> domainSurvivalRDD = surPairRDD.map(new Function<Tuple2<String,String>, DomainSurvival>() {

			@Override
			public DomainSurvival call(Tuple2<String, String> tuple){
				DomainSurvival survival = new DomainSurvival();
				try{
					String topDomain = tuple._1;
					String value = tuple._2;
					StringBuffer ipBuffer = new StringBuffer();
					StringBuffer aliveBuffer = new StringBuffer();
					if(StringUtils.isNotBlank(value)){
						String[] valueStr = value.split("@@");
						for(String str : valueStr){
							String[] ipAlive = str.split("@");
							String ip = ipAlive[0];
							String alive = ipAlive[1];
							
							//IP去重
							if(!ipBuffer.toString().contains(ip)){
								ipBuffer.append(ip).append("|");
								aliveBuffer.append(alive).append("|");
							}
						}
						survival.setDomain(topDomain);
						survival.setIp(getFilter(ipBuffer));
						survival.setIsAlive(getFilter(aliveBuffer));
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return survival;
			}
		});
		
		DataFrame domainSurvivalMergeFrame = sqlContext.createDataFrame(domainSurvivalRDD, DomainSurvival.class);
		
		return domainSurvivalMergeFrame;
	}

	@SuppressWarnings("serial")
	private static DataFrame geHistoryV1MergeFrame(SQLContext sqlContext,DataFrame domainHistoryV1Frame) {
		
		/**
		 *  +---------+--------------------+-------+-----------+----------+----------+
			|   domain|                  ip|isAlive|isEffective|join_count|timestamps|
			+---------+--------------------+-------+-----------+----------+----------+
			|apple.com|49.86.0.0|49.89.0.0 |      0|          0|         1|1492790553|
			+---------+--------------------+-------+-----------+----------+----------+
		 */
		JavaPairRDD<String,String> pairRDD = domainHistoryV1Frame.toJavaRDD().mapToPair(new PairFunction<Row,String,String>() {
			
			@Override
			public Tuple2<String,String> call(Row row){
				try{
					String domain = row.getAs("domain");
					String ip = row.getAs("ip");
					String timestamps = row.getAs("timestamps").toString();
					
					String topDomain = URLUtil.getDomainName("http://" + domain);
					
					if(StringUtils.isNotBlank(topDomain)){
						//将全量域名保存下来便于和天域名进行排重统计，计算接入数总量
						return new Tuple2<String,String>(topDomain,ip+"@"+timestamps+"@"+domain+"@"+row.getAs("isEffective"));
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>("","") ;
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2){
				if(StringUtils.isNotBlank(v1) && StringUtils.isNotBlank(v2)){
					return v1 + "@@" + v2;
				}
				return "";
			}
		});
		 //(apple.com,49.86.0.0|49.89.0.0@1492790553@apple.com)
		JavaRDD<DomainHistory> domainHistoryRDD = pairRDD.map(new Function<Tuple2<String,String>, DomainHistory>() {
			@Override
			public DomainHistory call(Tuple2<String, String> tuple){
				DomainHistory dh = new DomainHistory();
				try{
					String topDomain  = tuple._1;
					String value = tuple._2;
					StringBuffer ipBuffer = new StringBuffer();
					StringBuffer fullDomainBuffer = new StringBuffer();
					List<Long> timeList = new ArrayList<Long>();
					List<String> effectiveList = new ArrayList<String>();
					if(StringUtils.isNotBlank(value)){
						String[] valueStr = value.split("@@");
						for(String str : valueStr){
							String[] ipTimeDomain = str.split("@");
							String ips = ipTimeDomain[0];
							long time = Long.parseLong(ipTimeDomain[1]);
							String fullDomain = ipTimeDomain[2];
							String isEffective = ipTimeDomain[3];
							String[] ipStr = ips.split("\\|");
							timeList.add(time);
							for(String ip : ipStr){
								//IP去重
								if(!ipBuffer.toString().contains(ip)){
									ipBuffer.append(ip).append("|");
								}
							}
							fullDomainBuffer.append(fullDomain).append("|");
							if(!effectiveList.contains(isEffective)){
								effectiveList.add(isEffective);
							}
						}
						long effective = getFilterResult(effectiveList);
						
						dh.setDomain(topDomain);
						dh.setIp(getFilter(ipBuffer));
						dh.setTimestamps(Collections.min(timeList));//第一发现网站时间
						dh.setJoin_count(valueStr.length);//接入数,表示发现的该网站下的全量域名个数
						dh.setFullDomain(getFilter(fullDomainBuffer));
						dh.setIsEffective(effective);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return dh;
			}
		});
		
		DataFrame DomainHistoryV1Frame = sqlContext.createDataFrame(domainHistoryRDD, DomainHistory.class);
		
		return DomainHistoryV1Frame;
	}

	@SuppressWarnings("serial")
	private static void insertFullDomainToES(String esNode,final Broadcast<IPDatabase> ipBroadcast,final Broadcast<DomainDatabase> domainBroadcast,
			JavaRDD<DomainHistory> domainHistoryV3RDD,String date,SysProps sysProps){
		
		JavaRDD<FullDomainIndex> fullDomainRdd = domainHistoryV3RDD.map(new Function<DomainHistory,FullDomainIndex>() {

			@Override
			public FullDomainIndex call(DomainHistory his){
				FullDomainIndex index = new FullDomainIndex();
				try{
					String domain = his.getDomain();
					String ip = his.getIp();
					
					index.setYmid(0);
					
					index.setYm("p"+domain);
					//域名后缀
					String domainSuffix = CommonUtils.getDomainSuffix(domain);
					index.setYmSuffix(domainSuffix);
					
					//顶级域名
					String topDomain = URLUtil.getDomainName("http://" + domain);
					
					StructDomain structDomain = domainBroadcast.value().get(topDomain);
					//structDomain为null时，ES表所对应的字段值为null，不需要额外的设置
					if(structDomain != null){
						String site_name = structDomain.get(1);
						if(StringUtils.isNotBlank(site_name)){
							index.setSiteName("p"+site_name);
						}else{
							index.setSiteName("");
						}
						
						String registrar_name = structDomain.get(2);
						if(StringUtils.isNotBlank(registrar_name)){
							index.setYmzcsMc("p"+registrar_name);
						}else{
							index.setYmzcsMc("");
						}
						
						String recordState = structDomain.get(3);
						if(StringUtils.isNotBlank(recordState)){
							index.setIsRecord(recordState);
						}
						
						String cnDomainName = structDomain.get(4);
						if(StringUtils.isNotBlank(cnDomainName)){
							index.setDomainCN(cnDomainName);
						}
						
						//有数据为空字符;无数据时为null
						index.setHostUnit("");
						
						String licenseKey = structDomain.get(5);
						if(StringUtils.isNotBlank(licenseKey)){
							index.setLicense(licenseKey);
						}else{
							index.setLicense("");
						}
					}
					
					StringBuilder provinceBuffer = new StringBuilder();//省编码集合
					StringBuffer cityBuffer = new StringBuffer();//市编码集合
					StringBuffer countyBuffer = new StringBuffer();//县编码集合
					StringBuffer ispIdBuffer = new StringBuffer();//接入商Id集合
					StringBuffer ispNameBuffer = new StringBuffer();//接入商名称集合
					StringBuffer houseCodeBuffer = new StringBuffer();//机房编码集合
					StringBuffer houseNameBuffer = new StringBuffer();//机房名称集合
					StringBuffer ipBuffer = new StringBuffer();//IP集合
					StringBuffer idcIspIdBuffer = new StringBuffer();//接入商经营许可证号集合
					StringBuffer operatorsNameBuffer = new StringBuffer();//运营商名称集合
					
					if(StringUtils.isNotBlank(ip)){
						String[] ipSplit = ip.split("\\|");
						for(String ipStr : ipSplit){
							if(StringUtils.isNotBlank(ipStr)){
								StructIP structIp = ipBroadcast.value().get(ipStr);
								if(structIp == null){
									continue;
								}
								ipBuffer.append(ipStr).append(",");//IP集合
								provinceBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.PROVINCE)).append(",");//省编码集合
								cityBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.CITY)).append(",");//市编码集合
								countyBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.COUNTY)).append(",");//县编码集合
								ispIdBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_ID)).append(",");//接入商Id集合
								ispNameBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_NAME)).append(",");//接入商名称集合
								houseCodeBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.HOUSE_CODE)).append(",");//机房编码集合
								houseNameBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.HOUSE_NAME)).append(",");//机房名称集合
								idcIspIdBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.ISP_JYXKZBH)).append(",");//接入商经营许可证号集合
								operatorsNameBuffer.append(getDBValue(structIp,DnsStatConsts.IPExField.OPERATORS_NAME)).append(",");//运营商名称集合
							}
						}
					}
					String splitStr = ",";
					String[] ips = ipBuffer.toString().split(splitStr);
					index.setIps(ips);
					
					index.setShengIds(transformLongArray(provinceBuffer.toString(),splitStr,ips.length));
					index.setShiIds(transformLongArray(cityBuffer.toString(),splitStr,ips.length));
					index.setXianIds(transformLongArray(countyBuffer.toString(),splitStr,ips.length));
					
					long timestamps = his.getTimestamps();
					Date date = new Date(timestamps*1000);
					SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
					String fxsj = sdf.format(date);
					index.setFxsj(fxsj);
					
					index.setJrsIds(transformLongArray(ispIdBuffer.toString(),splitStr,ips.length));
					index.setJrsMcs(transformStringArray(ispNameBuffer.toString(),splitStr,ips.length));
					
					//网站级别:顶级域名算一级，在此基础上有几个点就加几级
					if(StringUtils.isNotBlank(topDomain)){
						if(StringUtils.equals(domain, topDomain)){
							index.setWzjb(1L);
						}else{
							index.setWzjb(Long.parseLong(CommonUtils.valueOf(domain.replace(topDomain,"").split("\\.").length))+1);
						}
						index.setDjym("p"+topDomain);
					}
					index.setSfch(his.getIsAlive());//存活状态
					index.setHouseCodes(transformStringArray(houseCodeBuffer.toString(),splitStr,ips.length));
					index.setHouseNames(transformStringArray(houseNameBuffer.toString(),splitStr,ips.length));
					index.setIdcIspIds(transformStringArray(idcIspIdBuffer.toString(),splitStr,ips.length));
					index.setYysMcs(transformStringArray(operatorsNameBuffer.toString(),splitStr,ips.length));
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return index;
			}
		});
		
		try{
			EsClientPool pool = null;
			EsClient esClient = null;
			int clientNum = sysProps.getInteger("es.clientNum", 1);
			
			//节点间交互的tcp端口
			String tcpPort = sysProps.getValue("transport.tcp.port");
			String clusterName = sysProps.getValue("es.clusterName");
			pool = new EsClientPool(clusterName,esNode, tcpPort, clientNum);
			esClient = pool.getEsClient();
		
			String dateStr = DateUtils.formatDateToES(DateUtils.stringToDate(date));
			String indexName = "dns_" + dateStr +"_ipdomainrelation";
			String type = "br2004";
			
			IndexClass indexClass = new IndexClass(indexName,type);
			EsEntity<FullDomainIndex> esEntity = new EsFullDomainVisits();
			EsMappingService es = new EsMappingServiceImpl(indexClass, pool, esEntity);
			
			// 先删除索引,避免数据重复录入同一索引
			if (esClient.isExists(indexName)) {
				esClient.deleteIndex(indexName);
				log.info("---------删除已存在的索引----------" + indexName);
			}
			String number_of_shards = sysProps.getValue("es.number_of_shards", "3");
			String number_of_replicas = sysProps.getValue("es.number_of_replicas", "0");
			es.init(number_of_shards,number_of_replicas);
			
			Map<String,String> map = new HashMap<String,String>();
			map.put("es.mapping.id","ym");
			JavaEsSpark.saveToEs(fullDomainRdd, "/"+indexName+"/"+type,map);
			
			//清除ES数据
			int keeyDate = sysProps.getInteger("esindex.keep.day",30); 
			deleteIndex(pool,"ipdomainrelation",keeyDate);
			
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}

	@SuppressWarnings("serial")
	private static void uploadDomainHistoryV3ToHDFS(JavaRDD<Row> domainHisV3RDD,SysProps sysProps) {
		
		JavaRDD<String> domainHistory3 = domainHisV3RDD.map(new Function<Row,String>() {

			@Override
			public String call(Row row) {
				StringBuffer buffer = new StringBuffer();
				try{
					String his_domain = row.getAs("his_domain");
					String his_ip = row.getAs("his_ip");
					String yd_domain = row.getAs("yd_domain");
					String yd_ip = row.getAs("yd_ip");
					
					if(StringUtils.isNotBlank(his_domain)){
						String timestamps = row.getAs("timestamps").toString();
						if(StringUtils.isNotBlank(yd_domain)){
							//天域名中的域名在新历史域名中存在，则合并IP，有效状态和存活状态不变
							buffer.append(his_domain+"\t");
							
							//天域名和历史域名有可能会有重复的IP,去掉重复的IP
							List<String> mergeIpList = new ArrayList<String>();
							mergeIpList.addAll(Arrays.asList(his_ip.split("\\|")));
							String[] ydIpStr = yd_ip.split("\\|");
							for(String str : ydIpStr){
								if(!mergeIpList.contains(str)){
									mergeIpList.add(str);
								}
							}
							
							//设置IP长度不超过10000
							int size = mergeIpList.size();
							if(size > 10000){
								size = 10000;
							}
							List<String> resultIp = mergeIpList.subList(0,size);
					    	StringBuffer bufferIp = new StringBuffer();
					    	for(String ip : resultIp){
					    		bufferIp.append(ip+"|");
					    	}
					    	
					    	buffer.append(getFilter(bufferIp)+"\t");
							buffer.append(timestamps+"\t");
							buffer.append(row.getAs("isEffective").toString()+"\t");
							buffer.append(row.getAs("isAlive").toString());
							
						}else{
							//历史域名在天域名没有找到对应关系
							buffer.append(his_domain+"\t");
							buffer.append(his_ip+"\t");
							buffer.append(timestamps+"\t");
							buffer.append(row.getAs("isEffective").toString()+"\t");
							buffer.append(row.getAs("isAlive").toString());
						}
					}else{
						//天域名在历史域名中没有找到对应关系,则直接新增，并将有效状态设置1，存活状态设置-1
						if(StringUtils.isNotBlank(yd_domain)){
							buffer.append(yd_domain+"\t");
							buffer.append(yd_ip+"\t");
							buffer.append(new Date().getTime()/1000+"\t");
							buffer.append("1"+"\t");
							buffer.append("-1");
						}
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		try{
			String domainPath = sysProps.getValue("hdfs.histroy.domain.dir");
			FileSystem fileSystem = FileSystem.get(new Configuration()); 
			Path path = new Path(domainPath);
			boolean flag = fileSystem.isDirectory(path);
			log.info("hdfspath=="+domainPath);
			log.info("flag=="+flag);
			if(flag){
				fileSystem.delete(path,true);
			}
			domainHistory3.saveAsTextFile(domainPath);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	/**
	 * 处理逻辑:1.若天域名在历史域名中没有对应关系，则新增该条记录，并设置存活状态为-1，有效状态为1
	 *        2.若历史域名中在天域名中没有对应关系，则该条记录保持不变
	 *        3.若找到对应关系，则合并IP，并去重
	 * @param sqlContext
	 * @param domainHisV3RDD
	 * @param surfilterIpList
	 * @return
	 */
	@SuppressWarnings("serial")
	private static JavaRDD<DomainHistory> getDomainHistroyV3RDD(SQLContext sqlContext,JavaRDD<Row> domainHisV3RDD,final List<String> surfilterIpList) {
		/**
			+-------------+--------------------+----------+-----------+-------+-------------+--------------------+
			|   his_domain|              his_ip|timestamps|isEffective|isAlive|    yd_domain|               yd_ip|
			+-------------+--------------------+----------+-----------+-------+-------------+--------------------+
			|  b.baidu.com|        36.152.8.242|1492790553|          1|     -1|         null|                null| 
			|         null|                null|      null|       null|   null|  a.apple.com|           49.86.0.0|
			|www.baidu.com| 49.81.0.0|49.87.0.0|1492790553|          1|      1|www.baidu.com|           49.82.0.0|
			+-------------+--------------------+----------+-----------+-------+-------------+--------------------+
		 */
		JavaRDD<DomainHistory> domainHistoryV3Rdd = domainHisV3RDD.map(new Function<Row,DomainHistory>() {

			@Override
			public DomainHistory call(Row row){
				DomainHistory domain = new DomainHistory();
				try{
					String his_domain = row.getAs("his_domain");
					String his_ip = row.getAs("his_ip");
					String yd_domain = row.getAs("yd_domain");
					String yd_ip = row.getAs("yd_ip");
					
					if(StringUtils.isNotBlank(his_domain)){
						String timestamps = row.getAs("timestamps").toString();
						if(StringUtils.isNotBlank(yd_domain)){
							//天域名中的域名在新历史域名中存在，则合并IP，有效状态和存活状态不变
							domain.setDomain(his_domain);
							domain.setIsAlive(Integer.parseInt(row.getAs("isAlive").toString()));
							domain.setIsEffective(Integer.parseInt(row.getAs("isEffective").toString()));
							domain.setTimestamps(Long.parseLong(timestamps));
							
							//天域名和历史域名有可能会有重复的IP,去掉重复的IP
							List<String> mergeIpList = new ArrayList<String>();
							mergeIpList.addAll(Arrays.asList(his_ip.split("\\|")));
							String[] ydIpStr = yd_ip.split("\\|");
							for(String str : ydIpStr){
								if(StringUtils.isNotBlank(str)){
									if(!mergeIpList.contains(str)){
										mergeIpList.add(str);
									}
								}
							}
							
							//设置IP长度不超过10000
							int size = mergeIpList.size();
							if(size > 10000){
								size = 10000;
							}
							List<String> resultIp = mergeIpList.subList(0,size);
					    	StringBuffer bufferIp = new StringBuffer();
					    	for(String ip : resultIp){
					    		bufferIp.append(ip+"|");
					    	}
					    	domain.setIp(getFilter(bufferIp));
						}else{
							//历史域名在天域名没有找到对应关系
							domain.setDomain(his_domain);
							domain.setIp(his_ip);
							domain.setTimestamps(Long.parseLong(timestamps));
							domain.setIsEffective(Integer.parseInt(row.getAs("isEffective").toString()));
							domain.setIsAlive(Integer.parseInt(row.getAs("isAlive").toString()));
						}
					}else{
						//天域名在历史域名中没有找到对应关系,则直接新增，并将有效状态设置1，存活状态设置-1
						if(StringUtils.isNotBlank(yd_domain)){
							domain.setDomain(yd_domain);
							domain.setIp(yd_ip);
							domain.setTimestamps(new Date().getTime()/1000);
							domain.setIsEffective(1);
							domain.setIsAlive(-1);
						}
						
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return domain;
			}
		});
		
		return domainHistoryV3Rdd;
	}
	/**
	 * 处理逻辑: 最终改变的是存活状态
	 *         1.新历史域名中V1中在存活中没有找到对应的关系，就将存活状态设置为-1，表示待检测
	 *         2.若找到对应关系，判断历史域名中的IP对应存活结果表中是否存在:
	 *           若存在，只要其对应的IP有一个是存活的，则这个域名对应的存活状态就是存活状态
	 *           若不存在，则同样将存活状态设置为-1                         
	 * @param sqlContext
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getDomainHistoryV2Frame(SQLContext sqlContext) {
		
		StringBuffer sql = new StringBuffer();
		sql.append("select history.domain,");
		sql.append("       history.ip          as history_ip,");
		sql.append("       history.timestamps,");
		sql.append("       history.isEffective,");
		sql.append("       history.isAlive	   as history_alive,");
		sql.append("       sur.ip              as survivalIp,");
		sql.append("       sur.isAlive		   as sur_alive");
		sql.append("  from domain_history_v1_temp history left join domain_survival_temp sur");
		sql.append("    on history.domain = sur.domain");
		
		DataFrame domainHisV2Frame = sqlContext.sql(sql.toString());
		
		/**
			+--------------+-------------------+----------+-----------+-------------+-----------+---------+
			|        domain|         history_ip|timestamps|isEffective|history_alive| survivalIp|sur_alive|
			+--------------+-------------------+----------+-----------+-------------+-----------+---------+
			|   b.baidu.com|        27.16.0.110|1499338529|         -1|           -1|27.16.0.110|        1|
			| www.baidu.com|27.16.0.0|27.16.0.1|1499338530|         -1|           -1|       null|     null|
			+--------------+-------------------+----------+-----------+-------------+-----------+---------+
		 */
		
		JavaPairRDD<String,String> domainHisV2Rdd = domainHisV2Frame.toJavaRDD().mapToPair(new PairFunction<Row,String, String>() {
			@Override
			public Tuple2<String,String> call(Row row){
				String key = "";
				String value = "";
				try{
					String domain = row.getAs("domain");
					String history_ip = row.getAs("history_ip");
					String timestamps = row.getAs("timestamps").toString();
					String isEffective = row.getAs("isEffective").toString();
					String history_alive = row.getAs("history_alive").toString();
					
					String survivalIp = row.getAs("survivalIp");
					
					if(StringUtils.isBlank(survivalIp)){
						value = "noData";
					}else{
						String sur_alive = row.getAs("sur_alive").toString();
						value = survivalIp + "@" + sur_alive;
					}
					key = domain + "@" + history_ip +"@" + timestamps + "@" +isEffective + "@" +history_alive;
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String,String>(key,value);
			}
		}).reduceByKey(new Function2<String, String, String>() {
			@Override
			public String call(String x, String y){
				return x + "@@" + y;
			}
		});
		//(b.baidu.com@27.16.0.110@1499338529@-1@-1,27.16.0.110@1),(www.baidu.com@27.16.0.0|27.16.0.1@1499338530@-1@-1,noData)
		JavaRDD<DomainHistory> domainHistoryV2Rdd = domainHisV2Rdd.map(new Function<Tuple2<String,String>, DomainHistory>() {

			@Override
			public DomainHistory call(Tuple2<String, String> tuple){
				DomainHistory dh = new DomainHistory();
				try{
					String key = tuple._1;
					String value = tuple._2;
					
					String[] keyStr = key.split("@");
					String historyIp = keyStr[1];

					dh.setDomain(keyStr[0]);
					dh.setIp(historyIp);
					dh.setTimestamps(Long.parseLong(keyStr[2]));
					dh.setIsEffective(Integer.parseInt(keyStr[3]));
					
					//将所有存活结果中的IP存入集合中
					ArrayList<String> surIpList = new ArrayList<String>();
					
					List<String> aliveList = new ArrayList<String>();
					//用于判断历史域名在存活结果中是否有对应记录
					boolean flag = false;
					String[] valueSplit = value.split("@@");
					if(valueSplit.length > 1){
						for(String ipAlive : valueSplit){
							if(!StringUtils.equals(ipAlive, "noData")){
								String[] ipAliveSplit = ipAlive.split("@");
								String surIp = ipAliveSplit[0];
								String alive = ipAliveSplit[1];
								
								if(!surIpList.contains(surIp)){
									surIpList.add(surIp);
								}
								if(!aliveList.contains(alive)){
									aliveList.add(alive);
								}
							}else{
								flag = true;
							}
						}
					}else{
						String ipAlive = valueSplit[0];
						if(!StringUtils.equals(ipAlive, "noData")){
							String[] ipAliveSplit = ipAlive.split("@");
							String surIp = ipAliveSplit[0];
							String alive = ipAliveSplit[1];
							
							if(!surIpList.contains(surIp)){
								surIpList.add(surIp);
							}
							if(!aliveList.contains(alive)){
								aliveList.add(alive);
							}
						}else{
							flag = true;
						}
					}
					if(flag){
						dh.setIsAlive(-1);//未找到域名对应IP存活关系，设置为-1表示待检测
					}else{
						//历史域名中的IP存入集合中
						ArrayList<String> historyIpList = new ArrayList<String>();
						historyIpList.addAll(Arrays.asList(keyStr[1].split("\\|")));
						
						//求交集,用于判断新历史域名对应IP是否在存活结果中
						historyIpList.retainAll(surIpList);
						
						if(historyIpList.size() > 0){//交集后的结果
							if(aliveList.contains("1")){
								dh.setIsAlive(1);
							}else{
								dh.setIsAlive(0);
							}
						}else{
							dh.setIsAlive(-1);//未找到域名对应IP存活关系，设置为-1表示待检测
						}
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return dh;
			}
		});
		
		//新历史域名V2.0
		DataFrame domainHistroyV2Frame = sqlContext.createDataFrame(domainHistoryV2Rdd, DomainHistory.class);
		return domainHistroyV2Frame;
	}
	/**
	 * 拨测结果中不满足条件的IP，只删IP，域名保留
	 * @param sqlContext
	 * @param hiveSql
	 * @param ipDatabase
	 * @param surfilterIpList
	 * @param date
	 * @param hiveDB
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getDomainInvalidFrame(SQLContext sqlContext,HiveContext hiveSql,final Broadcast<IPDatabase> ipBroadcast,final List<String> surfilterIpList,String date,String hiveDB){
		DataFrame aliveFrame = hiveSql.sql("select domain,ip from "+hiveDB+".domain_invalid where ds = '"+date+"'");
		
		JavaRDD<DomainInvalid> domainInvalidRDD = aliveFrame.toJavaRDD().map(new Function<Row, DomainInvalid>() {

			@Override
			public DomainInvalid call(Row row)  {
				DomainInvalid ds = new DomainInvalid();
				try{
					String domain = row.getAs("domain");
					String ip = row.getAs("ip");
					
					StringBuffer invalidIpBuffer = new StringBuffer();
					//拨测结果中IP过滤
					if(StringUtils.isNotBlank(ip)){
						String[] invalidIpStr = ip.split("\\|");
						for(String invIp : invalidIpStr){
							//过滤外省IP
							StructIP structSurIp = ipBroadcast.value().get(invIp);
							if(structSurIp != null){
								//114 IP过滤
								if(!surfilterIpList.contains(invIp)){
									invalidIpBuffer.append(invIp).append("|");
								}
							}
						}
					}
					ds.setIp(getFilter(invalidIpBuffer));
					ds.setDomain(domain);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return ds;
			}
		});
		
		DataFrame invalidFrame = sqlContext.createDataFrame(domainInvalidRDD, DomainInvalid.class);
		
		return invalidFrame;
	}
	
	/**
	 * 过滤省外IP和114IP，不满足条件的直接删除该记录
	 * @param sqlContext
	 * @param hiveSql
	 * @param ipDatabase
	 * @param surfilterIpList
	 * @param date
	 * @param hiveDB
	 * @return
	 */
	@SuppressWarnings("serial")
	private static DataFrame getDomainSurvivalFrame(SQLContext sqlContext,HiveContext hiveSql,final Broadcast<IPDatabase> ipBroadcast,final List<String> surfilterIpList,String date,String hiveDB){
		
		DataFrame aliveFrame = hiveSql.sql("select domain,ip,isAlive from "+hiveDB+".domain_survival where ds = '"+date+"'");
		
		JavaRDD<DomainSurvival> domainSurvivalRDD = aliveFrame.toJavaRDD().map(new Function<Row, DomainSurvival>() {

			@Override
			public DomainSurvival call(Row row)  {
				DomainSurvival ds = new DomainSurvival();
				try{
					String domain = row.getAs("domain");
					String ip = row.getAs("ip");
					String isAlive = row.getAs("isAlive").toString();
					
					//IP过滤
					if(StringUtils.isNotBlank(ip)){
						//过滤外省IP
						StructIP structSurIp = ipBroadcast.value().get(ip);
						if(structSurIp != null){//为null则为外省
							//114 IP过滤
							if(surfilterIpList.contains(ip)){
								ip = "";
							}
						}else{
							ip = "";
						}
					}
					ds.setDomain(domain);
					ds.setIp(ip);
					ds.setIsAlive(isAlive);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return ds;
			}
		});
		
		DataFrame survivalFrame = sqlContext.createDataFrame(domainSurvivalRDD, DomainSurvival.class);
		survivalFrame.registerTempTable("domain_survival_filter_temp");
		
		DataFrame survivalFrameResult = sqlContext.sql("select * from domain_survival_filter_temp where ip is not null and ip !=''");
		
		return survivalFrameResult;
	}
	
	@SuppressWarnings("serial")
	private static DataFrame getDomainIpDataFrame(HiveContext hiveSql,SQLContext sqlContext,DataFrame domainSurvivalFrame,String date,String hiveDB,final List<String> surfilterIpList) {
		//以拨测结果为准
		StringBuffer sql = new StringBuffer();
		sql.append("select inv.domain, inv.ip as invalidIp, sur.ip as survivalIp");
		sql.append("  from domain_invalid_temp inv");
		sql.append("  left join domain_survival_temp sur");
		sql.append("    on inv.domain = sur.domain");
		sql.append("   and sur.isAlive = 1");
		
		DataFrame domainIpFrame = hiveSql.sql(sql.toString());
		
		/**
		 *  +----------------+--------------------+----------+                              
			|          domain|           invalidIp|survivalIp|
			+----------------+--------------------+----------+
			|    www.sina.com|           49.85.0.0| 49.85.0.0|
			|     a.baidu.com|        36.152.8.240|      null|
			|www.baidu.com.cn|                null|      null|
			+----------------+--------------------+----------+
		 */
		JavaPairRDD<String,String> rdd = domainIpFrame.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() {

			@Override
			public Tuple2<String, String> call(Row row) throws Exception {
				String domain = row.getAs("domain");
				String invalidIp = row.getAs("invalidIp");
				String survivalIp = row.getAs("survivalIp");
				
				if(StringUtils.isBlank(invalidIp)){
					invalidIp = "N";
				}
				if(StringUtils.isBlank(survivalIp)){
					survivalIp = "N";
				}
				return new Tuple2<String,String>(domain+"@"+invalidIp,survivalIp);
			}
		}).reduceByKey(new Function2<String, String, String>() {

			@Override
			public String call(String v1, String v2) throws Exception {
				
				return v1 + "@@" + v2;
			}
		});
		
		JavaRDD<DomainIp> result = rdd.map(new Function<Tuple2<String,String>,DomainIp>() {

			@Override
			public DomainIp call(Tuple2<String, String> tuple){
				DomainIp di = new DomainIp();
				try{
					String key = tuple._1;
					String value = tuple._2;
					
					String[] keySplit = key.split("@");
					String invalidIp = keySplit[1];
					
					List<String> invalidIpList = new ArrayList<String>();
					StringBuffer ipBuffer = new StringBuffer();
					if(!StringUtils.equals(invalidIp, "N")){
						invalidIpList.addAll(Arrays.asList(invalidIp.split("\\|")));
						ipBuffer.append(invalidIp).append("|");
						String[] survivalIps = value.split("@@");
						for(String surIp : survivalIps){
							if(!StringUtils.equals(surIp, "N")){
								//IP去重
								if(!invalidIpList.contains(surIp)){
									invalidIpList.add(surIp);
									ipBuffer.append(surIp).append("|");
								}
							}
						}
					}else{
						String[] survivalIps = value.split("@@");
						for(String surIp : survivalIps){
							if(!StringUtils.equals(surIp, "N")){
								//IP去重
								if(!invalidIpList.contains(surIp)){
									invalidIpList.add(surIp);
									ipBuffer.append(surIp).append("|");
								}
							}
						}
					}
					
					di.setDomain(keySplit[0]);
					di.setIp(getFilter(ipBuffer));
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return di;
			}
		});
		
		DataFrame resultFrame = sqlContext.createDataFrame(result,DomainIp.class);
		
		return resultFrame;
	}
	
	private static String getDBValue(StructIP structIP,int filedID){
        String id = structIP.get(filedID);
        if (id.equals(DnsStatConsts.NA_STR)){
        	return "";
        }
        return id;
    }
	
	private static long parseLong(String str) {
		long data = 0;
		try {
			data = Long.parseLong(StringUtils.trim(str));
		} catch (Exception ex) {
		}
		return data;
	}
	
	private static Long[] transformLongArray(String source ,String splitStr,int arrSize){
		if (StringUtils.isBlank(splitStr) && arrSize > 0){
			return null;
		}
		Long[] arrReturn = new Long[arrSize];
		if (source != null ) {
			String[] arrStr = source.split(splitStr);
			for (int i = 0 ,strSize = arrStr.length; i < arrSize ; i++) {
				if (i < strSize) {
					try {
						arrReturn[i] = Long.valueOf( arrStr[i] );
					} catch (NumberFormatException e) {
						arrReturn[i] = 0l;
					}
				} else {
					arrReturn[i] = 0l;
				}
			}
		} else {
			for (int i = 0 ; i < arrSize ; i++) {
				arrReturn[i] = 0l;
			}
		}
		return arrReturn;
	}
	
	
	private static String[] transformStringArray(String source ,String splitStr , int arrSize){
		if (StringUtils.isBlank(splitStr) && arrSize > 0){
			return null;
		}
		String[] arrReturn = new String[arrSize];
		if (source != null ) {
			String[] arrStr = source.split(splitStr);
			for (int i = 0 ,strSize = arrStr.length; i < arrSize ; i++) {
				if (i < strSize) {
					arrReturn[i] = arrStr[i];
				} else {
					arrReturn[i] = "";
				}
			}
		} else {
			for (int i = 0 ; i < arrSize ; i++) {
				arrReturn[i] = "";
			}
		}
		return arrReturn;
	}
	
	private static void deleteIndex(EsClientPool pool,String type,int esIndexKeepDay){
    	try{
    		log.info("begin delete index: ");
        	EsClient esClient = pool.getEsClient();
        	List<String> lstDeleteIndex = deleteIndex(esIndexKeepDay,type, esClient);
    		for (String indexName : lstDeleteIndex) {
    			esClient.deleteIndex(indexName);
    			log.info("delete index: " + indexName);
    		}
    	}catch(Exception e){
    		if(log.isErrorEnabled()){
    			log.error(e.getMessage(),e);
    		}
    	}
    }
	
	public static List<String> deleteIndex(int keepDay, String type, EsClient esClient){
		MetaData metaData = esClient.getMetaData();
		ImmutableOpenMap<String, IndexMetaData> map = metaData.getIndices();
		map.keys();
		ObjectLookupContainer<String> keyContiner = map.keys();
		List<String> lstDeleteIndexName = new ArrayList<String>(16); 
		
		Calendar cal = Calendar.getInstance();
		cal.add(Calendar.DATE, -keepDay);
		Date baseDate = cal.getTime();
		log.info("baseDate:"+baseDate);
		for (ObjectCursor<String> key : keyContiner) {
			String dateStr = fetchIndexName(type, key.value);
			if (dateStr != null) {
				Date date = DateUtils.esToformatDate(dateStr);
				if (date != null && baseDate.after(date)) {
					lstDeleteIndexName.add(key.value);
				}
			}
		}
		return lstDeleteIndexName;
	}
	
	private static String fetchIndexName(String type, String str){
		String returnStr = null;
		if (str != null && type != null && str.contains(type) && str.length() > 14) {
			returnStr = str.substring(4, 14);
		}
		return returnStr;
	}
	
	public static Date getExecDate(Date date,String dayStr) {
		if (StringUtils.isNotBlank(dayStr)) {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
			try {
				date = sdf.parse(dayStr);
			} catch (ParseException e) {
				if(log.isErrorEnabled()){
					log.error(e.getMessage(),e);
				}
			}
		}
		return date;
	}
	
	public static String getFilter(StringBuffer buffer){
		String str = "";
		if(buffer.length() > 0){
			str = buffer.substring(0, buffer.length()-1);
		}
		return str;
	}
	
	private static long getFilterResult(List<String> list) {
		long result = 1;
		if(list.contains("1")){
			result = 1;
		}else if(list.contains("0")){
			result = 0;
		}else{
			result = -1;
		}
		return result;
	}
	
	private static List<String> getSurfilterIp(SysProps sysProps) {
		String surfilterIp = sysProps.getValue("surfilter.114.ips");
		List<String> surfilterIpList = new ArrayList<String>();
		if(StringUtils.isNotBlank(surfilterIp)){
			surfilterIpList = Arrays.asList(surfilterIp.split(","));
		}
		return surfilterIpList;
	}
}
